engine

type Bus ¶

type Bus struct {
	// contains filtered or unexported fields
}

Bus fans an event out to every attached sink. Emit is safe to call from multiple goroutines (parallel jobs emit concurrently); fan-out is serialized so sinks observe a single consistent ordering.

func NewBus(sinks ...Sink) *Bus

func (b *Bus) Emit(e Event)

type Event ¶

type Event struct {
	Type  EventType
	Time  time.Time
	RunID string
	// Seq is a per-run monotonic sequence number assigned by Bus.Emit. It lets
	// the event log and SSE replay order events and dedupe across a
	// replay-then-live handoff. Zero for events that never passed through a bus.
	Seq uint64

	// Job-scoped events.
	Job   string
	Stage string
	Exec  ExecKind

	// Group-scoped events.
	GroupID    string
	GroupKind  GroupKind
	GroupLabel string

	// RunStarted.
	Mode        RunMode
	HasMatrix   bool
	HasDetached bool
	ConfigPath  string
	ProjectPath string
	Commit      string // HEAD SHA at run start ("" outside a git repo)
	Branch      string // branch name at run start ("HEAD" when detached)

	// RunStarted / GroupStarted: job ordering for the board.
	Order []string

	// JobFinished / RunFinished.
	ExitCode int
	Duration time.Duration
	Err      string

	// LogLine / Diagnostic.
	Stream StreamKind
	Data   []byte
}

Event is a single item in the engine's output stream. Only the fields relevant to a given Type are populated.

type EventType ¶

type EventType int

EventType identifies what an Event reports. The engine emits a stream of these; sinks decide how to present or persist them.

const (
	RunStarted    EventType = iota // Mode, HasMatrix, HasDetached, Order
	RunFinished                    // Duration, Err
	GroupStarted                   // a concurrent barrier: parallel run / stage / matrix group (GroupKind, GroupLabel, Order)
	GroupFinished                  // GroupKind, GroupLabel
	JobStarted                     // Job, Stage, Exec, GroupID
	JobFinished                    // ExitCode, Duration, Err (also drives the status board)
	LogLine                        // Stream, Data (a chunk of job output)
	Diagnostic                     // Data (a fully-formatted internal log line)
)

type ExecKind ¶

type ExecKind int

ExecKind reports how a job is being run. It is an execution fact, not a presentation hint: the engine says whether a job runs alone, as part of a concurrent barrier, or detached, and the sink maps that to streaming vs a board vs inline status lines.

const (
	Standalone ExecKind = iota // sequential job, streamed on its own
	Concurrent                 // member of a concurrent barrier (a group)
	Detached                   // a `parallel: true` job detached from the chain
)

type GroupKind ¶

type GroupKind int

GroupKind distinguishes the three flavors of concurrent barrier so the sink can render the right header.

const (
	GroupParallelAll GroupKind = iota // whole-run parallel mode (no per-group header)
	GroupStage                        // a stage in parallel-stages mode ("Stage: X")
	GroupMatrix                       // a matrix group inside a sequential run ("Matrix [X]")
)

type JobExecutor ¶

type JobExecutor interface {
	Execute(ctx context.Context, job config.JobConfig, out io.Writer) error
}

JobExecutor runs a single job to completion, writing all of the job's output (image pull progress and container logs) to out. Internal diagnostics are emitted through the logger injected into the concrete implementation, not through out. Returning a non-nil error means the job failed.

The interface exists so the engine can run without a Docker daemon in tests (via a fake) and so other front-ends can reuse the engine unchanged.

type JobState ¶

type JobState int

JobState is the lifecycle state shown on the status board.

const (
	StatePending JobState = iota
	StateRunning
	StatePassed
	StateFailed
)

type RunMode ¶

type RunMode int

RunMode selects how the prepared jobs are scheduled.

const (
	ModeSequential RunMode = iota
	ModeParallel
	ModeParallelStages
)

type Runner ¶

type Runner struct {
	// contains filtered or unexported fields
}

Runner schedules and executes the prepared jobs, emitting events to the bus. It owns no presentation: stdout, log files, and the status board live in the sinks attached to the bus.

func (r *Runner) Cleanup(ctx context.Context) error

func (r *Runner) PrepareJobConfigs(options RunnerOptions) error

func (r *Runner) Run(executor JobExecutor) error

type RunnerOptions ¶

type RunnerOptions struct {
	// contains filtered or unexported fields
}

type Sink ¶

type Sink interface {
	Emit(Event)
}

Sink consumes events. Implementations must not call back into the Bus they are attached to (Emit holds the bus lock while fanning out).

type Spec ¶

type Spec struct {
	ConfigFile string
	JobNames   []string
	Stages     []string
	Remote     string
	Env        []string
	Mode       RunMode
}

Spec describes a single pipeline run request. It is the engine's input, built by a front-end (the CLI today, the server later) from its own flags or API payload.

type StreamKind ¶

type StreamKind int

StreamKind identifies which output stream a LogLine came from. Today the engine merges everything into Stdout (matching the previous behavior); the field exists so later phases can split streams without a schema change.

const (
	StreamStdout StreamKind = iota
	StreamStderr
)

type WireEvent ¶ added in v0.2.0

type WireEvent struct {
	Seq         uint64    `json:"seq"`
	Type        string    `json:"type"`
	Time        time.Time `json:"time"`
	RunID       string    `json:"runId,omitempty"`
	Job         string    `json:"job,omitempty"`
	Stage       string    `json:"stage,omitempty"`
	Exec        string    `json:"exec,omitempty"`
	GroupID     string    `json:"groupId,omitempty"`
	GroupKind   string    `json:"groupKind,omitempty"`
	GroupLabel  string    `json:"groupLabel,omitempty"`
	Mode        string    `json:"mode,omitempty"`
	HasMatrix   bool      `json:"hasMatrix,omitempty"`
	HasDetached bool      `json:"hasDetached,omitempty"`
	Order       []string  `json:"order,omitempty"`
	ConfigPath  string    `json:"configPath,omitempty"`
	ProjectPath string    `json:"projectPath,omitempty"`
	Commit      string    `json:"commit,omitempty"`
	Branch      string    `json:"branch,omitempty"`
	ExitCode    int       `json:"exitCode,omitempty"`
	DurationMs  int64     `json:"durationMs,omitempty"`
	Err         string    `json:"err,omitempty"`
	Stream      string    `json:"stream,omitempty"`
	Data        string    `json:"data,omitempty"`
}

WireEvent is the JSON representation of an Event, shared by the event-log sink and the server's SSE stream so both emit an identical, self-describing shape (string-named enums, type-specific fields only). It is the contract the UI consumes.

func ToWire(e Event) WireEvent