feat(P03): task execution engine with HCL specs, jobs, tasks, WaitDelay

Implements Phase 3 of v0.1 Foundation: - internal/model/job.go: Job + Task models with status state machines - internal/store/migrations/0002_jobs_tasks.sql: jobs + tasks tables with FK - internal/store/job_task_repo.go: JobRepo + TaskRepo with CRUD and lifecycle updates - internal/jobspec/spec.go: HCL parser using hashicorp/hcl/v2 hclsimple - internal/jobspec/spec_test.go: 4 tests for parser - internal/engine/executor.go: parallel task executor using os/exec with Go 1.25 WaitDelay for clean process shutdown - internal/cli/job.go: orca job {run,list,stop,logs} wired to executor - testdata/hello.hcl, testdata/fail.hcl: smoke test fixtures Verified: job run executes commands, captures stdout/stderr, persists state, job stop transitions status, job logs displays captured output. All tests pass with -race. ---ci--- project: orca phase: 3 milestone: v0.1 status: execute req_covered: - REQ-004 - REQ-006 - REQ-009 - REQ-018 - REQ-020 - REQ-021 ---/ci---
2026-06-03 12:45:20 +00:00
parent 78334f1f74
commit f9a9873341
11 changed files with 812 additions and 28 deletions
@@ -1,9 +1,18 @@
 package cli

 import (
+	"context"
+	"errors"
 	"fmt"
+	"time"

+	"github.com/google/uuid"
 	"github.com/spf13/cobra"
+
+	"git.cloudinit.dev/coreci/orca/internal/engine"
+	"git.cloudinit.dev/coreci/orca/internal/jobspec"
+	"git.cloudinit.dev/coreci/orca/internal/model"
+	"git.cloudinit.dev/coreci/orca/internal/store"
 )

 var jobCmd = &cobra.Command{
@@ -12,46 +21,187 @@ var jobCmd = &cobra.Command{
 	Long:  "Run, list, stop, and inspect orca jobs.",
 }

+func jobExecutor() (*engine.Executor, func() error, error) {
+	db, closer, err := openDB()
+	if err != nil {
+		return nil, nil, err
+	}
+	jobs := store.NewJobRepo(db)
+	tasks := store.NewTaskRepo(db)
+	return engine.NewExecutor(jobs, tasks, newLogger()), closer, nil
+}
+
 var jobRunCmd = &cobra.Command{
 	Use:   "run <spec.hcl>",
 	Short: "Run a job from an HCL spec file",
-	Long:  "Submit a job spec and execute it. Implemented in Phase 3.",
+	Long:  "Submit a job spec, execute its tasks, and persist the result.",
 	Args:  cobra.ExactArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
-		return notImplemented("orca job run " + args[0])
+		spec, err := jobspec.ParseFile(args[0])
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
+		defer cancel()
+
+		exec, closer, err := jobExecutor()
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		job := &model.Job{
+			ID:     uuid.NewString(),
+			Name:   spec.Job.Name,
+			Spec:   args[0],
+			Status: model.JobStatusPending,
+		}
+		if err := exec.Run(ctx, job, toTaskSpecs(spec.Tasks)); err != nil {
+			if jsonOutput {
+				_ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": err.Error()})
+				return err
+			}
+			fmt.Fprintf(cmd.ErrOrStderr(), "✗ Job %s failed: %v\n", job.ID, err)
+			return err
+		}
+		if jsonOutput {
+			return printJSON(map[string]any{"id": job.ID, "name": job.Name, "status": "complete"})
+		}
+		fmt.Fprintf(cmd.OutOrStdout(), "✓ Job complete: %s (%s)\n", job.ID, job.Name)
+		return nil
 	},
 }

 var jobListCmd = &cobra.Command{
 	Use:   "list",
 	Short: "List all jobs",
-	Long:  "Display all jobs and their status. Implemented in Phase 3.",
+	Long:  "Display all jobs and their status.",
 	RunE: func(cmd *cobra.Command, args []string) error {
-		return notImplemented("orca job list")
+		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
+		defer cancel()
+
+		db, closer, err := openDB()
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		jobs, err := store.NewJobRepo(db).List(ctx)
+		if err != nil {
+			return err
+		}
+		if jsonOutput {
+			return printJSON(jobs)
+		}
+		if len(jobs) == 0 {
+			fmt.Fprintln(cmd.OutOrStdout(), "No jobs. Use 'orca job run <spec.hcl>' to submit one.")
+			return nil
+		}
+		fmt.Fprintf(cmd.OutOrStdout(), "%-36s  %-20s  %-12s  %-8s\n", "ID", "NAME", "STATUS", "EXIT")
+		for _, j := range jobs {
+			fmt.Fprintf(cmd.OutOrStdout(), "%-36s  %-20s  %-12s  %-8d\n", j.ID, j.Name, j.Status, j.ExitCode)
+		}
+		return nil
 	},
 }

+var (
+	stopID string
+)
+
 var jobStopCmd = &cobra.Command{
-	Use:   "stop <job-id>",
+	Use:   "stop [job-id]",
 	Short: "Stop a running job",
-	Long:  "Stop a job by ID. Implemented in Phase 3.",
-	Args:  cobra.ExactArgs(1),
+	Long:  "Mark a job as stopped. Note: this is a soft stop (cancel context for the daemon).",
+	Args:  cobra.MaximumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
-		return notImplemented("orca job stop " + args[0])
+		id := stopID
+		if id == "" && len(args) > 0 {
+			id = args[0]
+		}
+		if id == "" {
+			return fmt.Errorf("job id required (--id or argument)")
+		}
+		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
+		defer cancel()
+
+		db, closer, err := openDB()
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		repo := store.NewJobRepo(db)
+		job, err := repo.Get(ctx, id)
+		if err != nil {
+			if errors.Is(err, store.ErrNotFound) {
+				return fmt.Errorf("job not found: %s", id)
+			}
+			return err
+		}
+		if err := repo.UpdateStatus(ctx, id, model.JobStatusStopped, 130); err != nil {
+			return err
+		}
+		if jsonOutput {
+			return printJSON(map[string]any{"id": id, "status": "stopped", "previous_status": job.Status})
+		}
+		fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s\n", id)
+		return nil
 	},
 }

 var jobLogsCmd = &cobra.Command{
-	Use:   "logs <job-id>",
-	Short: "Show logs for a job",
-	Long:  "Display the logs for a job by ID. Implemented in Phase 3.",
-	Args:  cobra.ExactArgs(1),
+	Use:   "logs [job-id]",
+	Short: "Show task output for a job",
+	Long:  "Display captured stdout/stderr for all tasks in a job.",
+	Args:  cobra.MaximumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
-		return notImplemented("orca job logs " + args[0])
+		id := stopID
+		if id == "" && len(args) > 0 {
+			id = args[0]
+		}
+		if id == "" {
+			return fmt.Errorf("job id required (--id or argument)")
+		}
+		ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
+		defer cancel()
+
+		db, closer, err := openDB()
+		if err != nil {
+			return err
+		}
+		defer closer()
+
+		taskRepo := store.NewTaskRepo(db)
+		tasks, err := taskRepo.ListByJob(ctx, id)
+		if err != nil {
+			return err
+		}
+		if jsonOutput {
+			return printJSON(tasks)
+		}
+		if len(tasks) == 0 {
+			fmt.Fprintln(cmd.OutOrStdout(), "No tasks for this job.")
+			return nil
+		}
+		for i, t := range tasks {
+			fmt.Fprintf(cmd.OutOrStdout(), "--- task[%d] %s (%s) exit=%d ---\n", i, t.Command, t.Status, t.ExitCode)
+			if t.Stdout != "" {
+				fmt.Fprintln(cmd.OutOrStdout(), t.Stdout)
+			}
+			if t.Stderr != "" {
+				fmt.Fprintln(cmd.OutOrStderr(), t.Stderr)
+			}
+		}
+		return nil
 	},
 }

 func init() {
+	jobStopCmd.Flags().StringVar(&stopID, "id", "", "job id")
+	jobLogsCmd.Flags().StringVar(&stopID, "id", "", "job id")
+
 	jobCmd.AddCommand(jobRunCmd)
 	jobCmd.AddCommand(jobListCmd)
 	jobCmd.AddCommand(jobStopCmd)
@@ -59,16 +209,15 @@ func init() {
 	rootCmd.AddCommand(jobCmd)
 }

-func notImplemented(cmd string) error {
-	if jsonOutput {
-		return printJSON(map[string]any{
-			"command": cmd,
-			"status":  "not_implemented",
-			"phase":   "1-cli-skeleton",
-			"next":    "Phase 2-6 will implement this",
-		})
+func toTaskSpecs(in []jobspec.TaskSpec) []engine.TaskSpec {
+	out := make([]engine.TaskSpec, len(in))
+	for i, t := range in {
+		out[i] = engine.TaskSpec{
+			Name:    t.Name,
+			Command: t.Command,
+			Args:    t.Args,
+			Env:     t.Env,
+		}
 	}
-	fmt.Fprintf(rootCmd.ErrOrStderr(), "✗ %s: not yet implemented (Phase 1: CLI skeleton only)\n", cmd)
-	fmt.Fprintf(rootCmd.ErrOrStderr(), "  see .ciagent/ROADMAP.md for the full 6-phase plan\n")
-	return fmt.Errorf("not implemented: %s", cmd)
+	return out
 }