package workflow import ( "bytes" "context" "encoding/json" "fmt" "maps" "sort" "strings" "time" "gopkg.in/yaml.v3" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/sandbox" "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/models" ) type Action struct { Uses string Path string Using string Main string Inputs map[string]string Source []byte } type ActionResolver func(uses string) (*Action, error) type Options struct { Sandbox sandbox.Runner ResolveAction ActionResolver MaxConsoleBytes int64 MaxOutputBytes int64 } type JobResult struct { Status string Outputs map[string]any Error string } type Context struct { Inputs map[string]any Env map[string]any Secrets map[string]string Outputs map[string]map[string]any Annotations map[string][]models.Annotation Cache models.Cache Vars map[string]any JobResults map[string]JobResult Console []string Shell string Error error console int64 } type Executor struct { opts Options Context *Context jobFailed bool jobDeadline time.Time } func New(options Options) *Executor { return &Executor{ opts: options, Context: &Context{ Inputs: map[string]any{}, Env: map[string]any{}, Secrets: map[string]string{}, Outputs: map[string]map[string]any{}, Annotations: map[string][]models.Annotation{}, Cache: models.Cache{}, Vars: map[string]any{}, JobResults: map[string]JobResult{}, Shell: "bash -xe", }, } } func Parse(raw json.RawMessage) (*models.Workflow, error) { trimmed := bytes.TrimSpace(raw) if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) { return nil, fmt.Errorf("workflow is required") } var text string if err := json.Unmarshal(trimmed, &text); err == nil { text = strings.TrimSpace(text) if text == "" || text == "null" { return nil, fmt.Errorf("workflow is required") } var parsed models.Workflow if err := yaml.Unmarshal([]byte(text), &parsed); err != nil { return nil, fmt.Errorf("failed to parse workflow YAML: %w", err) } return &parsed, nil } var parsed models.Workflow if err := yaml.Unmarshal(trimmed, &parsed); err != nil { return nil, fmt.Errorf("failed to parse workflow: %w", err) } return &parsed, nil } func HasRunStep(raw json.RawMessage) (string, bool) { parsed, err := Parse(raw) if err != nil { return "", false } names := make([]string, 0, len(parsed.Jobs)) for name := range parsed.Jobs { names = append(names, name) } sort.Strings(names) for _, name := range names { for _, step := range parsed.Jobs[name].Steps { if strings.TrimSpace(step.Run) != "" { return name, true } } } return "", false } func (e *Executor) Execute(ctx context.Context, request *models.Request) (*models.Result, error) { if request == nil { return nil, fmt.Errorf("request is required") } workflow, err := Parse(request.Workflow) if err != nil { return nil, err } if err := e.initializeContext(request); err != nil { return nil, err } if workflow.Name != "" { e.Context.Env["GITHUB_WORKFLOW"] = workflow.Name } for key, value := range workflow.Env { evaluated, evalErr := e.evaluateExpression(ctx, fmt.Sprintf("%v", value)) if evalErr != nil { return e.failureResult(evalErr), nil } e.Context.Env[key] = evaluated } order, err := jobOrder(workflow) if err != nil { return e.failureResult(err), nil } var firstErr error for _, name := range order { job := workflow.Jobs[name] result, jobErr := e.executeJob(ctx, name, &job) e.Context.JobResults[name] = result if jobErr != nil && firstErr == nil { firstErr = jobErr } } if firstErr != nil { e.Context.Error = firstErr return e.failureResult(firstErr), nil } return e.successResult(), nil } func (e *Executor) initializeContext(request *models.Request) error { maps.Copy(e.Context.Inputs, request.Inputs) if request.Context != nil { if config, ok := request.Context["config"].(map[string]any); ok { if env, ok := config["env"].(map[string]any); ok { maps.Copy(e.Context.Env, env) } } if secrets, ok := request.Context["secrets"].(map[string]any); ok { for key, value := range secrets { if text, ok := value.(string); ok { e.Context.Secrets[key] = text } } } if cacheValue, ok := request.Context["cache"]; ok && cacheValue != nil { e.Context.Cache = decodeCache(cacheValue) } if vars, ok := request.Context["vars"].(map[string]any); ok { e.Context.Vars = vars } } if token := e.Context.Secrets["GITHUB_TOKEN"]; token != "" { if _, exists := e.Context.Env["GITHUB_TOKEN"]; !exists { e.Context.Env["GITHUB_TOKEN"] = token } } e.Context.Env["CI"] = "true" e.Context.Env["GITHUB_ACTIONS"] = "true" return nil } func jobOrder(workflow *models.Workflow) ([]string, error) { names := make([]string, 0, len(workflow.Jobs)) for name := range workflow.Jobs { names = append(names, name) } sort.Strings(names) visited := map[string]bool{} visiting := map[string]bool{} order := make([]string, 0, len(names)) var visit func(string) error visit = func(name string) error { if visited[name] { return nil } if visiting[name] { return fmt.Errorf("circular needs involving job %s", name) } job, ok := workflow.Jobs[name] if !ok { return fmt.Errorf("job %s is not defined", name) } visiting[name] = true for _, need := range needsOf(job.Needs) { if err := visit(need); err != nil { return err } } visiting[name] = false visited[name] = true order = append(order, name) return nil } for _, name := range names { if err := visit(name); err != nil { return nil, err } } return order, nil } func needsOf(value any) []string { switch typed := value.(type) { case string: if typed == "" { return nil } return []string{typed} case []string: return typed case []any: out := make([]string, 0, len(typed)) for _, item := range typed { if name, ok := item.(string); ok && name != "" { out = append(out, name) } } return out default: return nil } } func (e *Executor) executeJob(ctx context.Context, name string, job *models.WorkflowJob) (JobResult, error) { result := JobResult{Status: "success", Outputs: map[string]any{}} dependencyFailed := false for _, need := range needsOf(job.Needs) { dep, ok := e.Context.JobResults[need] if ok && (dep.Status == "failure" || dep.Status == "skipped") { dependencyFailed = true } } e.jobFailed = dependencyFailed if job.If != nil { guard, err := e.evaluateGuard(ctx, job.If) if err != nil { result.Status = "failure" result.Error = err.Error() return result, fmt.Errorf("job %s: %w", name, err) } if !guard.run { result.Status = "skipped" return result, nil } } else if dependencyFailed { result.Status = "skipped" return result, nil } jobCtx := ctx if job.TimeoutMinutes > 0 { var cancel context.CancelFunc jobCtx, cancel = context.WithTimeout(ctx, time.Duration(job.TimeoutMinutes)*time.Minute) defer cancel() } e.jobDeadline = time.Time{} if job.TimeoutMinutes > 0 { e.jobDeadline = time.Now().Add(time.Duration(job.TimeoutMinutes) * time.Minute) } envSnapshot := make(map[string]any, len(e.Context.Env)) maps.Copy(envSnapshot, e.Context.Env) for key, value := range job.Env { evaluated, evalErr := e.evaluateExpression(ctx, fmt.Sprintf("%v", value)) if evalErr != nil { maps.Copy(e.Context.Env, envSnapshot) result.Status = "failure" result.Error = evalErr.Error() return result, fmt.Errorf("job %s: %w", name, evalErr) } e.Context.Env[key] = evaluated } err := e.executeSteps(jobCtx, job) outputs := map[string]any{} for outputName, expression := range job.Outputs { value, evalErr := e.evaluateExpression(ctx, fmt.Sprintf("%v", expression)) if evalErr != nil { outputs[outputName] = "" continue } outputs[outputName] = value } for key := range e.Context.Env { if _, existed := envSnapshot[key]; !existed { delete(e.Context.Env, key) } } maps.Copy(e.Context.Env, envSnapshot) result.Outputs = outputs if err != nil { result.Status = "failure" result.Error = err.Error() return result, fmt.Errorf("job %s: %w", name, err) } return result, nil } func (e *Executor) executeSteps(ctx context.Context, job *models.WorkflowJob) error { e.jobFailed = false e.Context.Error = nil var firstErr error for index, step := range job.Steps { if err := ctx.Err(); err != nil { e.jobFailed = true return err } stepName := fmt.Sprintf("step_%d", index+1) if step.ID != "" { stepName = step.ID } else if step.Name != "" { stepName = step.Name } guard, err := e.evaluateGuard(ctx, step.If) if err != nil { firstErr = fmt.Errorf("step %s: %w", stepName, err) e.jobFailed = true break } if e.jobFailed && !guard.always && !guard.failure { continue } if !guard.run { continue } if step.Shell != "" { e.Context.Shell = step.Shell } env, err := e.buildStepEnv(ctx, &step) if err != nil { firstErr = fmt.Errorf("step %s: %w", stepName, err) e.jobFailed = true break } e.emit(fmt.Sprintf("##[group]%s", stepName)) var stepErr error if strings.TrimSpace(step.Run) != "" { stepErr = fmt.Errorf("this evaluator runs gha-lite policies net-only and refuses a run: step (job step %s)", stepName) } else if step.Uses != "" { stepErr = e.executeStepUses(ctx, &step, env) } e.emit("##[endgroup]") if stepErr != nil { e.emit(fmt.Sprintf("##[error]step %s failed: %v", stepName, stepErr)) if step.ContinueOnError { continue } e.jobFailed = true e.Context.Error = stepErr if firstErr == nil { firstErr = fmt.Errorf("step %s: %w", stepName, stepErr) } continue } } if firstErr != nil { return firstErr } if !e.jobDeadline.IsZero() && time.Now().After(e.jobDeadline) { e.jobFailed = true return fmt.Errorf("job exceeded timeout-minutes") } return nil } func (e *Executor) emit(line string) { if e.opts.MaxConsoleBytes > 0 { if e.Context.console >= e.opts.MaxConsoleBytes { return } remaining := e.opts.MaxConsoleBytes - e.Context.console if int64(len(line)) > remaining { line = line[:remaining] } } e.Context.Console = append(e.Context.Console, line) e.Context.console += int64(len(line)) + 1 } func (e *Executor) successResult() *models.Result { outputs := models.Outputs{} for name, result := range e.Context.JobResults { outputs[name] = result.Outputs } if used, limit := e.outputBytes(outputs); limit > 0 && used > limit { err := fmt.Errorf("workflow outputs exceed the configured limit") e.emit("##[error]" + err.Error()) return e.failureResult(err) } return &models.Result{ Status: models.ExecutionComplete, ExitStatus: models.ExitSuccess, Outputs: outputs, Cache: e.Context.Cache, Annotations: e.annotationMap(), ConsoleOutput: strings.Join(e.Context.Console, "\n"), } } func (e *Executor) failureResult(err error) *models.Result { annotations := e.annotationMap() errors := []models.Annotation{} if err != nil { errors = append(errors, models.Annotation{AnnotationLevel: models.AnnotationError, Message: err.Error(), Title: err.Error()}) } annotations["error"] = errors return &models.Result{ Status: models.ExecutionComplete, ExitStatus: models.ExitFailure, Outputs: models.Outputs{}, Cache: e.Context.Cache, Annotations: annotations, ConsoleOutput: strings.Join(e.Context.Console, "\n"), } } func (e *Executor) annotationMap() models.Annotations { out := models.Annotations{} for level, list := range e.Context.Annotations { copied := make([]models.Annotation, len(list)) copy(copied, list) out[level] = copied } return out } func (e *Executor) outputBytes(outputs models.Outputs) (int64, int64) { if e.opts.MaxOutputBytes <= 0 { return 0, e.opts.MaxOutputBytes } data, err := json.Marshal(outputs) if err != nil { return 0, e.opts.MaxOutputBytes } return int64(len(data)), e.opts.MaxOutputBytes } func decodeCache(value any) models.Cache { data, err := json.Marshal(value) if err != nil { return models.Cache{} } var out models.Cache if err := json.Unmarshal(data, &out); err != nil { return models.Cache{} } if out == nil { out = models.Cache{} } return out }