tenseleyflow/shithub / 732831b

+// SPDX-License-Identifier: AGPL-3.0-or-later

+

+// Command shithubd-runner executes shithub Actions workflow jobs.

+package main

+

+func main() {

+	Execute()

+}

+// SPDX-License-Identifier: AGPL-3.0-or-later

+

+package main

+

+import (

+	"fmt"

+	"os"

+

+	"github.com/spf13/cobra"

+)

+

+var rootCmd = &cobra.Command{

+	Use:           "shithubd-runner",

+	Short:         "Run shithub Actions jobs on a Docker or Podman host",

+	SilenceUsage:  true,

+	SilenceErrors: true,

+}

+

+func Execute() {

+	if err := rootCmd.Execute(); err != nil {

+		fmt.Fprintln(os.Stderr, "shithubd-runner:", err)

+		os.Exit(1)

+	}

+}

+

+func init() {

+	rootCmd.AddCommand(runCmd)

+	rootCmd.AddCommand(versionCmd)

+}

+// SPDX-License-Identifier: AGPL-3.0-or-later

+

+package main

+

+import (

+	"os"

+	"time"

+

+	"github.com/spf13/cobra"

+	"github.com/spf13/pflag"

+

+	infralog "github.com/tenseleyFlow/shithub/internal/infra/log"

+	runnerpkg "github.com/tenseleyFlow/shithub/internal/runner"

+	runnerapi "github.com/tenseleyFlow/shithub/internal/runner/api"

+	runnerconfig "github.com/tenseleyFlow/shithub/internal/runner/config"

+	"github.com/tenseleyFlow/shithub/internal/runner/engine"

+	"github.com/tenseleyFlow/shithub/internal/runner/workspace"

+)

+

+var runConfigPath string

+

+var runCmd = &cobra.Command{

+	Use:   "run",

+	Short: "Claim and execute shithub Actions jobs",

+	RunE: func(cmd *cobra.Command, _ []string) error {

+		cfg, err := runnerconfig.Load(runnerconfig.LoadOptions{

+			ConfigPath: runConfigPath,

+			Overrides:  flagOverrides(cmd),

+		})

+		if err != nil {

+			return err

+		}

+		logger := infralog.New(infralog.Options{

+			Level:  cfg.Log.Level,

+			Format: cfg.Log.Format,

+			Writer: os.Stderr,

+		}).With("component", "runner")

+

+		workspaces := workspace.New(cfg.Runner.WorkspaceRoot)

+		removed, err := workspaces.Sweep(cfg.Runner.WorkspaceTTL, time.Now().UTC())

+		if err != nil {

+			return err

+		}

+		if removed > 0 {

+			logger.InfoContext(cmd.Context(), "swept stale workspaces", "count", removed)

+		}

+

+		client, err := runnerapi.New(runnerapi.Config{

+			BaseURL:     cfg.Server.BaseURL,

+			RunnerToken: cfg.Runner.Token,

+		})

+		if err != nil {

+			return err

+		}

+		execEngine := engine.NewDocker(engine.DockerConfig{

+			Binary:       cfg.Engine.Kind,

+			DefaultImage: cfg.Engine.DefaultImage,

+			Network:      cfg.Engine.Network,

+			Memory:       cfg.Engine.Memory,

+			CPUs:         cfg.Engine.CPUs,

+			Stdout:       os.Stdout,

+			Stderr:       os.Stderr,

+		})

+		r := runnerpkg.New(runnerpkg.Options{

+			API:          client,

+			Engine:       execEngine,

+			Workspaces:   workspaces,

+			Logger:       logger,

+			Labels:       cfg.Runner.Labels,

+			Capacity:     cfg.Runner.Capacity,

+			PollInterval: cfg.Runner.PollInterval,

+			DefaultImage: cfg.Engine.DefaultImage,

+			Clock:        func() time.Time { return time.Now().UTC() },

+		})

+		return r.Run(cmd.Context())

+	},

+}

+

+func init() {

+	runCmd.Flags().StringVar(&runConfigPath, "config", "", "Path to runner config file")

+	runCmd.Flags().String("server-url", "", "shithub base URL")

+	runCmd.Flags().String("token", "", "Runner registration token")

+	runCmd.Flags().String("labels", "", "Comma-separated runner labels")

+	runCmd.Flags().Int("capacity", 0, "Maximum concurrent jobs this runner advertises")

+	runCmd.Flags().Duration("poll-interval", 0, "Idle heartbeat interval")

+	runCmd.Flags().String("workspace-root", "", "Workspace root directory")

+	runCmd.Flags().Duration("workspace-ttl", 0, "Startup sweep TTL for stale workspaces")

+	runCmd.Flags().String("engine", "", "Execution engine: docker or podman")

+	runCmd.Flags().String("image", "", "Default container image")

+	runCmd.Flags().String("network", "", "Container network")

+	runCmd.Flags().String("memory", "", "Container memory limit")

+	runCmd.Flags().String("cpus", "", "Container CPU limit")

+	runCmd.Flags().String("log-level", "", "Log level: debug, info, warn, error")

+	runCmd.Flags().String("log-format", "", "Log format: text or json")

+}

+

+func flagOverrides(cmd *cobra.Command) map[string]string {

+	keys := map[string]string{

+		"server-url":     "server.base_url",

+		"token":          "runner.token",

+		"labels":         "runner.labels",

+		"capacity":       "runner.capacity",

+		"poll-interval":  "runner.poll_interval",

+		"workspace-root": "runner.workspace_root",

+		"workspace-ttl":  "runner.workspace_ttl",

+		"engine":         "engine.kind",

+		"image":          "engine.default_image",

+		"network":        "engine.network",

+		"memory":         "engine.memory",

+		"cpus":           "engine.cpus",

+		"log-level":      "log.level",

+		"log-format":     "log.format",

+	}

+	out := make(map[string]string)

+	cmd.Flags().Visit(func(f *pflag.Flag) {

+		if key, ok := keys[f.Name]; ok {

+			out[key] = f.Value.String()

+		}

+	})

+	return out

+}

+// SPDX-License-Identifier: AGPL-3.0-or-later

+

+package main

+

+import (

+	"fmt"

+	"runtime"

+

+	"github.com/spf13/cobra"

+

+	"github.com/tenseleyFlow/shithub/internal/version"

+)

+

+var versionCmd = &cobra.Command{

+	Use:   "version",

+	Short: "Print version, commit, build time, and Go runtime",

+	Run: func(_ *cobra.Command, _ []string) {

+		fmt.Printf("shithubd-runner %s\n", version.Version)

+		fmt.Printf("  commit: %s\n", version.Commit)

+		fmt.Printf("  built:  %s\n", version.BuiltAt)

+		fmt.Printf("  go:     %s\n", runtime.Version())

+		fmt.Printf("  os:     %s/%s\n", runtime.GOOS, runtime.GOARCH)

+	},

+}

+// SPDX-License-Identifier: AGPL-3.0-or-later

+

+// Package runner orchestrates the shithubd-runner claim/execute/status loop.

+package runner

+

+import (

+	"context"

+	"errors"

+	"fmt"

+	"io"

+	"log/slog"

+	"time"

+

+	"github.com/tenseleyFlow/shithub/internal/runner/api"

+	"github.com/tenseleyFlow/shithub/internal/runner/engine"

+)

+

+type API interface {

+	Heartbeat(ctx context.Context, req api.HeartbeatRequest) (*api.Claim, error)

+	UpdateStatus(ctx context.Context, jobID int64, token string, req api.StatusRequest) (api.StatusResponse, error)

+}

+

+type Workspaces interface {

+	Prepare(runID, jobID int64) (string, error)

+	Remove(runID, jobID int64) error

+}

+

+type SleepFunc func(ctx context.Context, d time.Duration) error

+

+type Options struct {

+	API          API

+	Engine       engine.Engine

+	Workspaces   Workspaces

+	Logger       *slog.Logger

+	Labels       []string

+	Capacity     int

+	PollInterval time.Duration

+	DefaultImage string

+	Clock        func() time.Time

+	Sleep        SleepFunc

+}

+

+type Runner struct {

+	api          API

+	engine       engine.Engine

+	workspaces   Workspaces

+	logger       *slog.Logger

+	labels       []string

+	capacity     int

+	pollInterval time.Duration

+	defaultImage string

+	clock        func() time.Time

+	sleep        SleepFunc

+}

+

+func New(opts Options) *Runner {

+	logger := opts.Logger

+	if logger == nil {

+		logger = slog.New(slog.NewTextHandler(io.Discard, nil))

+	}

+	clock := opts.Clock

+	if clock == nil {

+		clock = func() time.Time { return time.Now().UTC() }

+	}

+	sleep := opts.Sleep

+	if sleep == nil {

+		sleep = defaultSleep

+	}

+	poll := opts.PollInterval

+	if poll <= 0 {

+		poll = 5 * time.Second

+	}

+	capacity := opts.Capacity

+	if capacity <= 0 {

+		capacity = 1

+	}

+	return &Runner{

+		api:          opts.API,

+		engine:       opts.Engine,

+		workspaces:   opts.Workspaces,

+		logger:       logger,

+		labels:       append([]string{}, opts.Labels...),

+		capacity:     capacity,

+		pollInterval: poll,

+		defaultImage: opts.DefaultImage,

+		clock:        clock,

+		sleep:        sleep,

+	}

+}

+

+func (r *Runner) Run(ctx context.Context) error {

+	for {

+		claimed, err := r.RunOnce(ctx)

+		if err != nil {

+			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {

+				return err

+			}

+			r.logger.ErrorContext(ctx, "runner loop iteration failed", "error", err)

+		}

+		if claimed {

+			continue

+		}

+		if err := r.sleep(ctx, r.pollInterval); err != nil {

+			return err

+		}

+	}

+}

+

+func (r *Runner) RunOnce(ctx context.Context) (bool, error) {

+	claim, err := r.api.Heartbeat(ctx, api.HeartbeatRequest{Labels: r.labels, Capacity: r.capacity})

+	if err != nil {

+		return false, err

+	}

+	if claim == nil {

+		return false, nil

+	}

+	token := claim.Token

+	started := r.clock()

+	workspaceDir, err := r.workspaces.Prepare(claim.Job.RunID, claim.Job.ID)

+	if err != nil {

+		statusErr := r.complete(ctx, claim.Job.ID, token, engine.ConclusionFailure, started, r.clock())

+		return true, errors.Join(fmt.Errorf("prepare workspace: %w", err), statusErr)

+	}

+	defer func() {

+		if err := r.workspaces.Remove(claim.Job.RunID, claim.Job.ID); err != nil {

+			r.logger.WarnContext(ctx, "workspace cleanup failed", "run_id", claim.Job.RunID, "job_id", claim.Job.ID, "error", err)

+		}

+	}()

+

+	running, err := r.api.UpdateStatus(ctx, claim.Job.ID, token, api.StatusRequest{

+		Status:    "running",

+		StartedAt: started,

+	})

+	if err != nil {

+		return true, fmt.Errorf("mark job running: %w", err)

+	}

+	if running.NextToken == "" {

+		return true, errors.New("mark job running: server did not return next_token")

+	}

+	token = running.NextToken

+

+	outcome, execErr := r.engine.Execute(ctx, toEngineJob(claim.Job, workspaceDir, r.defaultImage))

+	conclusion := outcome.Conclusion

+	if conclusion == "" {

+		conclusion = engine.ConclusionFailure

+	}

+	completed := outcome.CompletedAt

+	if completed.IsZero() {

+		completed = r.clock()

+	}

+	if outcome.StartedAt.IsZero() {

+		outcome.StartedAt = started

+	}

+	if err := r.complete(ctx, claim.Job.ID, token, conclusion, outcome.StartedAt, completed); err != nil {

+		return true, err

+	}

+	if execErr != nil {

+		r.logger.WarnContext(ctx, "job completed with failing engine outcome", "job_id", claim.Job.ID, "conclusion", conclusion, "error", execErr)

+	}

+	return true, nil

+}

+

+func (r *Runner) complete(ctx context.Context, jobID int64, token, conclusion string, started, completed time.Time) error {

+	_, err := r.api.UpdateStatus(ctx, jobID, token, api.StatusRequest{

+		Status:      "completed",

+		Conclusion:  conclusion,

+		StartedAt:   started,

+		CompletedAt: completed,

+	})

+	if err != nil {

+		return fmt.Errorf("mark job completed: %w", err)

+	}

+	return nil

+}

+

+func toEngineJob(job api.Job, workspaceDir, defaultImage string) engine.Job {

+	steps := make([]engine.Step, 0, len(job.Steps))

+	for _, step := range job.Steps {

+		steps = append(steps, engine.Step{

+			ID:               step.ID,

+			Index:            step.Index,

+			StepID:           step.StepID,

+			Name:             step.Name,

+			If:               step.If,

+			Run:              step.Run,

+			Uses:             step.Uses,

+			WorkingDirectory: step.WorkingDirectory,

+			Env:              step.Env,

+			With:             step.With,

+			ContinueOnError:  step.ContinueOnError,

+		})

+	}

+	return engine.Job{

+		ID:             job.ID,

+		RunID:          job.RunID,

+		RepoID:         job.RepoID,

+		RunIndex:       job.RunIndex,

+		WorkflowFile:   job.WorkflowFile,

+		WorkflowName:   job.WorkflowName,

+		HeadSHA:        job.HeadSHA,

+		HeadRef:        job.HeadRef,

+		Event:          job.Event,

+		JobKey:         job.JobKey,

+		JobName:        job.JobName,

+		RunsOn:         job.RunsOn,

+		Needs:          append([]string{}, job.Needs...),

+		If:             job.If,

+		TimeoutMinutes: job.TimeoutMinutes,

+		Permissions:    job.Permissions,

+		Env:            job.Env,

+		Steps:          steps,

+		WorkspaceDir:   workspaceDir,

+		Image:          defaultImage,

+	}

+}

+

+func defaultSleep(ctx context.Context, d time.Duration) error {

+	timer := time.NewTimer(d)

+	defer timer.Stop()

+	select {

+	case <-ctx.Done():

+		return ctx.Err()

+	case <-timer.C:

+		return nil

+	}

+}

+// SPDX-License-Identifier: AGPL-3.0-or-later

+

+package runner

+

+import (

+	"context"

+	"errors"

+	"testing"

+	"time"

+

+	"github.com/tenseleyFlow/shithub/internal/runner/api"

+	"github.com/tenseleyFlow/shithub/internal/runner/engine"

+)

+

+type fakeAPI struct {

+	claim    *api.Claim

+	statuses []api.StatusRequest

+	tokens   []string

+}

+

+func (f *fakeAPI) Heartbeat(_ context.Context, _ api.HeartbeatRequest) (*api.Claim, error) {

+	return f.claim, nil

+}

+

+func (f *fakeAPI) UpdateStatus(_ context.Context, _ int64, token string, req api.StatusRequest) (api.StatusResponse, error) {

+	f.tokens = append(f.tokens, token)

+	f.statuses = append(f.statuses, req)

+	if req.Status == "running" {

+		return api.StatusResponse{NextToken: "next-token"}, nil

+	}

+	return api.StatusResponse{}, nil

+}

+

+type fakeEngine struct {

+	job engine.Job

+	out engine.Outcome

+	err error

+}

+

+func (f *fakeEngine) Execute(_ context.Context, job engine.Job) (engine.Outcome, error) {

+	f.job = job

+	return f.out, f.err

+}

+

+func (f *fakeEngine) StreamLogs(_ context.Context, _ int64) (<-chan engine.LogChunk, error) {

+	ch := make(chan engine.LogChunk)

+	close(ch)

+	return ch, nil

+}

+

+func (f *fakeEngine) Cancel(_ context.Context, _ int64) error { return nil }

+

+type fakeWorkspaces struct {

+	dir     string

+	removed bool

+	err     error

+}

+

+func (f *fakeWorkspaces) Prepare(_, _ int64) (string, error) {

+	return f.dir, f.err

+}

+

+func (f *fakeWorkspaces) Remove(_, _ int64) error {

+	f.removed = true

+	return nil

+}

+

+func TestRunOnce_NoClaim(t *testing.T) {

+	t.Parallel()

+	r := New(Options{API: &fakeAPI{}, Engine: &fakeEngine{}, Workspaces: &fakeWorkspaces{}})

+	claimed, err := r.RunOnce(t.Context())

+	if err != nil {

+		t.Fatalf("RunOnce: %v", err)

+	}

+	if claimed {

+		t.Fatal("claimed = true")

+	}

+}

+

+func TestRunOnce_ExecutesAndCompletesSuccess(t *testing.T) {

+	t.Parallel()

+	now := time.Date(2026, 5, 10, 21, 0, 0, 0, time.UTC)

+	claim := &api.Claim{

+		Token: "job-token",

+		Job: api.Job{

+			ID:    10,

+			RunID: 20,

+			Env:   map[string]string{"A": "B"},

+			Steps: []api.Step{{ID: 30, Run: "echo hi"}},

+		},

+	}

+	fapi := &fakeAPI{claim: claim}

+	fengine := &fakeEngine{out: engine.Outcome{Conclusion: engine.ConclusionSuccess, StartedAt: now, CompletedAt: now.Add(time.Second)}}

+	workspaces := &fakeWorkspaces{dir: "/tmp/workspace"}

+	r := New(Options{

+		API:          fapi,

+		Engine:       fengine,

+		Workspaces:   workspaces,

+		DefaultImage: "runner-image",

+		Clock:        func() time.Time { return now },

+	})

+	claimed, err := r.RunOnce(t.Context())

+	if err != nil {

+		t.Fatalf("RunOnce: %v", err)

+	}

+	if !claimed {

+		t.Fatal("claimed = false")

+	}

+	if fengine.job.WorkspaceDir != "/tmp/workspace" || fengine.job.Image != "runner-image" {

+		t.Fatalf("engine job: %#v", fengine.job)

+	}

+	if len(fapi.statuses) != 2 {

+		t.Fatalf("statuses: %#v", fapi.statuses)

+	}

+	if fapi.statuses[0].Status != "running" || fapi.statuses[1].Conclusion != engine.ConclusionSuccess {

+		t.Fatalf("statuses: %#v", fapi.statuses)

+	}

+	if fapi.tokens[0] != "job-token" || fapi.tokens[1] != "next-token" {

+		t.Fatalf("tokens: %#v", fapi.tokens)

+	}

+	if !workspaces.removed {

+		t.Fatal("workspace was not removed")

+	}

+}

+

+func TestRunOnce_EngineFailureStillCompletesJob(t *testing.T) {

+	t.Parallel()

+	now := time.Date(2026, 5, 10, 21, 0, 0, 0, time.UTC)

+	fapi := &fakeAPI{claim: &api.Claim{Token: "job-token", Job: api.Job{ID: 10, RunID: 20}}}

+	fengine := &fakeEngine{

+		out: engine.Outcome{Conclusion: engine.ConclusionFailure, StartedAt: now, CompletedAt: now.Add(time.Second)},

+		err: errors.New("exit 1"),

+	}

+	r := New(Options{

+		API:        fapi,

+		Engine:     fengine,

+		Workspaces: &fakeWorkspaces{dir: "/tmp/workspace"},

+		Clock:      func() time.Time { return now },

+	})

+	if _, err := r.RunOnce(t.Context()); err != nil {

+		t.Fatalf("RunOnce: %v", err)

+	}

+	if fapi.statuses[1].Conclusion != engine.ConclusionFailure {

+		t.Fatalf("completion: %#v", fapi.statuses[1])

+	}

+}

+

+func TestRunOnce_PrepareFailureMarksJobFailed(t *testing.T) {

+	t.Parallel()

+	fapi := &fakeAPI{claim: &api.Claim{Token: "job-token", Job: api.Job{ID: 10, RunID: 20}}}

+	r := New(Options{

+		API:        fapi,

+		Engine:     &fakeEngine{},

+		Workspaces: &fakeWorkspaces{err: errors.New("disk full")},

+		Clock:      func() time.Time { return time.Date(2026, 5, 10, 21, 0, 0, 0, time.UTC) },

+	})

+	claimed, err := r.RunOnce(t.Context())

+	if err == nil {

+		t.Fatal("RunOnce returned nil error")

+	}

+	if !claimed {

+		t.Fatal("claimed = false")

+	}

+	if len(fapi.statuses) != 1 || fapi.statuses[0].Conclusion != engine.ConclusionFailure {

+		t.Fatalf("statuses: %#v", fapi.statuses)

+	}

+}