Commit 8394af3

Eric Bower  ·  2026-08-08 11:54:27 -0400 EDT
parent 957187e
feat: add local runner base command with isolated execution and html logging

- Default `pici [destination]` command copies CWD to /tmp for isolated execution
- Streamline CLI flags with -e/--env for key-value environment overrides
- Isolate local sessions with `local.` prefix so `pici monitor` daemons ignore local dev runs
- Auto-detect git commit SHA and branch for attestation.json and event metadata
- Rsync build artifacts directly into a `<job_id>` subfolder at target destination
- Add onboarding guide when pico.sh is missing and root `pici help` overview
2 files changed,  +500, -36
+396, -32
......@@ -45,6 +45,17 @@ func defaultWorkspaceFactory(cfg *Cfg, logger *slog.Logger, source string) Works
4545 }
4646 }
4747
48+type envList []string
49+
50+func (e *envList) String() string {
51+ return strings.Join(*e, ", ")
52+}
53+
54+func (e *envList) Set(value string) error {
55+ *e = append(*e, value)
56+ return nil
57+}
58+
4859 type Cfg struct {
4960 Logger *slog.Logger
5061 Ctx context.Context
......@@ -61,6 +72,7 @@ type Cfg struct {
6172 IncludeRunning bool // emit running status updates in addition to terminal
6273 HumanOutput bool // human-readable output instead of JSONL / slog
6374 Wait bool // block until job completes, print history and summary
75+ EnvVars envList // custom environment variables passed via -e / -env
6476 }
6577
6678 type Event struct {
......@@ -80,6 +92,7 @@ func NewCfg() (*Cfg, string, bool) {
8092 var monitorInterval time.Duration
8193 var gcInterval time.Duration
8294 var logLevel string
95+ var envVars envList
8396 flag.StringVar(&keyLoc, "pk", "", "ssh private key used to authenticate with pico services")
8497 flag.StringVar(&certLoc, "ck", "", "ssh certificate public key used to authenticate with pico services (only required if using ssh certificates)")
8598 flag.StringVar(&artifactDir, "artifact-dir", "/tmp/pici-artifacts", "local directory to stage artifacts")
......@@ -87,6 +100,8 @@ func NewCfg() (*Cfg, string, bool) {
87100 flag.DurationVar(&monitorInterval, "monitor-interval", 5*time.Second, "interval for monitoring zmx sessions")
88101 flag.DurationVar(&gcInterval, "gc-interval", 10*time.Minute, "interval for garbage collection in monitor (0 to disable)")
89102 flag.StringVar(&logLevel, "log-level", "info", "log level: debug, info, warn, error")
103+ flag.Var(&envVars, "e", "environment variable in KEY=VAL format (can be specified multiple times)")
104+ flag.Var(&envVars, "env", "environment variable in KEY=VAL format (can be specified multiple times)")
90105 var includeRunning bool
91106 var human bool
92107 var wait bool
......@@ -118,20 +133,30 @@ func NewCfg() (*Cfg, string, bool) {
118133 IncludeRunning: includeRunning,
119134 HumanOutput: human,
120135 Wait: wait,
136+ EnvVars: envVars,
121137 }, cmd, wantHelp
122138 }
123139
140+func isKnownSubcommand(s string) bool {
141+ switch s {
142+ case "runner", "monitor", "cancel", "gc", "status", "orca", "run", "help":
143+ return true
144+ default:
145+ return false
146+ }
147+}
148+
124149 // splitCommand separates the first non-flag argument (the subcommand) from
125150 // the rest of the flags, so "runner --wait" becomes flags=["--wait"], cmd="runner".
126151 // It strips --help/help so we can print custom help per subcommand.
127152 func splitCommand(args []string) (flags []string, cmd string, wantHelp bool) {
128153 flags = make([]string, 0, len(args))
129154 for _, arg := range args {
130- if arg == "--help" || arg == "help" {
155+ if arg == "--help" || arg == "help" || arg == "-h" {
131156 wantHelp = true
132157 continue
133158 }
134- if cmd == "" && arg != "" && !strings.HasPrefix(arg, "-") {
159+ if cmd == "" && isKnownSubcommand(arg) {
135160 cmd = arg
136161 } else {
137162 flags = append(flags, arg)
......@@ -160,17 +185,67 @@ func newLogger(space string, levelStr string) *slog.Logger {
160185 })).With("service", space)
161186 }
162187
188+func printMainHelp() {
189+ fmt.Println(`pici — minimal parallel CI runner & monitor powered by zmx
190+
191+LOCAL DEVELOPER USAGE
192+ pici [destination] [flags] Run ./pico.sh locally in /tmp & render HTML logs
193+ pici pgs.sh:/my-site Run locally and rsync HTML logs to destination
194+ pici run [destination] Explicit alias for local run
195+
196+DAEMON & SERVICE COMMANDS
197+ pici runner Execute CI job from event JSON payload (stdin/flag)
198+ pici monitor Poll ci.* zmx sessions & stage/sync HTML artifacts
199+ pici cancel Cancel active running jobs for a repository
200+ pici gc Clean up stale/finished zmx sessions & artifacts
201+
202+SUBCOMMAND HELP
203+ pici <command> --help Show detailed help for a specific command (e.g. pici runner --help)
204+
205+FLAGS
206+ -e, -env <KEY=VAL> Set or override environment variable for pico.sh
207+ -pk <path> SSH private key
208+ -ck <path> SSH public certificate key
209+ -artifact-dir <path> Artifact staging directory (default: /tmp/pici-artifacts)
210+ -log-level <level> Log level: debug, info, warn, error`)
211+}
212+
213+func printMissingPicoHelp(cwd string) {
214+ fmt.Printf(`❌ Error: no pico.sh found in %s
215+
216+To run local CI tasks, create a pico.sh script in your project root:
217+
218+ #!/usr/bin/env bash
219+ set -euo pipefail
220+
221+ # Run parallel steps using zmx
222+ zmx run lint -d <your lint command>
223+ zmx run test -d <your test command>
224+
225+ # Wait for all steps to finish
226+ zmx wait "*"
227+ printf "\x1b[32msuccess!\x1b[0m\n"
228+
229+For full command documentation and daemon options, run:
230+ pici help
231+
232+`, cwd)
233+}
234+
163235 func main() {
164236 cfg, cmd, wantHelp := NewCfg()
165237
166238 cfg.Logger.Debug("setting up ci", "cfg", cfg)
167239 cfg.Logger.Debug("running cmd", "cmd", cmd)
168240
169- if wantHelp && (cmd == "runner" || cmd == "monitor") {
170- if cmd == "runner" {
241+ if wantHelp {
242+ switch cmd {
243+ case "runner":
171244 printRunnerHelp()
172- } else {
245+ case "monitor":
173246 printMonitorHelp()
247+ default:
248+ printMainHelp()
174249 }
175250 return
176251 }
......@@ -204,9 +279,23 @@ func main() {
204279 cfg.Logger.Debug("starting status updater")
205280 case "orca":
206281 cfg.Logger.Debug("starting orchestrator")
282+ case "help":
283+ printMainHelp()
284+ case "run", "":
285+ dest := ""
286+ if flag.NArg() > 0 {
287+ dest = flag.Arg(0)
288+ }
289+ if err := runLocal(cfg, dest); err != nil {
290+ cfg.Logger.Error("local run failed", "err", err)
291+ os.Exit(1)
292+ }
207293 default:
208- cfg.Logger.Error("must provide command: runner, cancel, gc, monitor, status, or orca")
209- os.Exit(1)
294+ dest := cmd
295+ if err := runLocal(cfg, dest); err != nil {
296+ cfg.Logger.Error("local run failed", "err", err)
297+ os.Exit(1)
298+ }
210299 }
211300 }
212301
......@@ -362,7 +451,9 @@ func (w *WorkspaceRsync) Setup() error {
362451 }
363452
364453 func (w *WorkspaceRsync) Cleanup() error {
365- // return os.RemoveAll(w.Dest)
454+ if w.Dest != "" {
455+ return os.RemoveAll(w.Dest)
456+ }
366457 return nil
367458 }
368459
......@@ -522,9 +613,13 @@ func (eng *JobEngine) Setup() error {
522613 }
523614
524615 func (eng *JobEngine) Run(manifest string) error {
525- prefix := fmt.Sprintf("ci.%s.%s.", eng.Ev.Name, eng.JobID)
616+ domain := "ci"
617+ if eng.Ev != nil && eng.Ev.Type == "local" {
618+ domain = "local"
619+ }
620+ prefix := fmt.Sprintf("%s.%s.%s.", domain, eng.Ev.Name, eng.JobID)
526621 // Child sessions use ".step." sub-prefix so zmx wait "*" inside pico.sh
527- // matches ci.<name>.<jobID>.step.* but NOT ci.<name>.<jobID>.runner.
622+ // matches <domain>.<name>.<jobID>.step.* but NOT <domain>.<name>.<jobID>.runner.
528623 // This avoids a deadlock where the runner waits for itself.
529624 childPrefix := prefix + "step."
530625
......@@ -551,9 +646,8 @@ func (eng *JobEngine) Run(manifest string) error {
551646 if eng.Ev.Tag != "" {
552647 cmdEnv = append(cmdEnv, fmt.Sprintf("PICI_TAG=%s", eng.Ev.Tag))
553648 }
554-
555- zmxPrefixStr := fmt.Sprintf("ZMX_SESSION_PREFIX=%s", childPrefix)
556- cmd := exec.Command("zmx", "run", runnerName, "-d", zmxPrefixStr, "bash", manifest)
649+ bashCmd := fmt.Sprintf("export ZMX_SESSION_PREFIX=%q; exec bash %q", childPrefix, manifest)
650+ cmd := exec.Command("zmx", "run", runnerName, "-d", "bash", "-c", bashCmd)
557651 cmd.Env = cmdEnv
558652 cmd.Dir = eng.Wk.GetDir()
559653
......@@ -610,14 +704,18 @@ func eventHandler(cfg *Cfg, eventData *Event) error {
610704 Ev: eventData,
611705 JobID: jobID,
612706 }
707+ var runErr error
613708 defer func() {
614- if err := eng.Cleanup(); err != nil {
615- cfg.Logger.Error("engine cleanup", "err", err)
709+ if runErr != nil || cfg.Wait {
710+ if err := eng.Cleanup(); err != nil {
711+ cfg.Logger.Error("engine cleanup", "err", err)
712+ }
616713 }
617714 }()
618715
619716 fmt.Fprintf(os.Stdout, "📦 syncing workspace %s\n", eventData.Workspace) //nolint:errcheck
620717 if err := eng.Setup(); err != nil {
718+ runErr = err
621719 return fmt.Errorf("setup: %w", err)
622720 }
623721 fmt.Fprintf(os.Stdout, "✅ workspace ready %s\n", eng.Wk.GetDir()) //nolint:errcheck
......@@ -654,6 +752,7 @@ func eventHandler(cfg *Cfg, eventData *Event) error {
654752
655753 manifest, err := eng.FindManifest()
656754 if err != nil {
755+ runErr = err
657756 fmt.Fprintf(os.Stdout, "❌ %s\n\n", err) //nolint:errcheck
658757 //nolint:errcheck
659758 fmt.Fprint(os.Stdout, `Create a pico.sh script in your workspace root:
......@@ -686,13 +785,15 @@ See: https://github.com/picosh/pici
686785
687786 fmt.Fprint(os.Stdout, "🏃 launching sessions...\n") //nolint:errcheck
688787 if err := eng.Run(manifest); err != nil {
788+ runErr = err
689789 return fmt.Errorf("run: %w", err)
690790 }
691791
692792 fmt.Fprintln(os.Stdout, "✅ job launched") //nolint:errcheck
693793
694794 if cfg.Wait {
695- if err := waitAndReport(cfg, log, eventData.Name, jobID); err != nil {
795+ if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
796+ runErr = err
696797 return fmt.Errorf("wait: %w", err)
697798 }
698799 return nil
......@@ -707,9 +808,17 @@ See: https://github.com/picosh/pici
707808
708809 // waitAndReport polls the job's sessions until all complete, prints live
709810 // progress to stdout, then dumps session history and a final summary.
710-func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID string) error {
711- prefix := "ci." + name + "." + jobID + "."
712- ticker := time.NewTicker(cfg.MonitorInterval)
811+func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) error {
812+ domain := "ci"
813+ if eventType == "local" {
814+ domain = "local"
815+ }
816+ prefix := domain + "." + name + "." + jobID + "."
817+ interval := cfg.MonitorInterval
818+ if interval <= 0 {
819+ interval = 5 * time.Second
820+ }
821+ ticker := time.NewTicker(interval)
713822 defer ticker.Stop()
714823
715824 // Handle ^C gracefully
......@@ -726,9 +835,16 @@ func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID string) error {
726835 var sessionOrder []string // track insertion order for deterministic output
727836 var liveLines []string // last set of status lines printed (for overwrite)
728837
838+ var done <-chan struct{}
839+ if cfg.Ctx != nil {
840+ done = cfg.Ctx.Done()
841+ }
842+
729843 for {
730844 select {
731- case <-cfg.Ctx.Done():
845+ case <-done:
846+ cancelJobSessions(prefix)
847+ fmt.Fprintln(os.Stdout, "\n⏹ job cancelled") //nolint:errcheck
732848 return cfg.Ctx.Err()
733849 case <-sigCh:
734850 fmt.Fprintln(os.Stdout, "\n⏹ cancelled") //nolint:errcheck
......@@ -876,9 +992,11 @@ func cleanSessionShort(name, prefix, repoName, jobID string) string {
876992 short := strings.TrimPrefix(name, prefix)
877993 // Strip "step." prefix added by child sessions
878994 short = strings.TrimPrefix(short, "step.")
879- // Strip nested full prefix (e.g. runner named ci.name.jobID.runner)
880- nested := "ci." + repoName + "." + jobID + "."
881- short = strings.TrimPrefix(short, nested)
995+ // Strip nested full prefix (e.g. runner named ci.name.jobID.runner or local.name.jobID.runner)
996+ nestedCI := "ci." + repoName + "." + jobID + "."
997+ short = strings.TrimPrefix(short, nestedCI)
998+ nestedLocal := "local." + repoName + "." + jobID + "."
999+ short = strings.TrimPrefix(short, nestedLocal)
8821000 return short
8831001 }
8841002
......@@ -917,10 +1035,11 @@ func runMonitor(cfg *Cfg) error {
9171035 defer ticker.Stop()
9181036
9191037 // Optional GC ticker — runs garbage collection on a separate interval.
920- var gcTicker *time.Ticker
1038+ var gcChan <-chan time.Time
9211039 if cfg.GCInterval > 0 {
922- gcTicker = time.NewTicker(cfg.GCInterval)
1040+ gcTicker := time.NewTicker(cfg.GCInterval)
9231041 defer gcTicker.Stop()
1042+ gcChan = gcTicker.C
9241043 }
9251044
9261045 // Track per-job display state across ticks (for human output)
......@@ -938,7 +1057,7 @@ func runMonitor(cfg *Cfg) error {
9381057 if err := monitorTick(cfg, log, output, jobStates); err != nil {
9391058 log.Error("monitor tick", "err", err)
9401059 }
941- case <-gcTicker.C:
1060+ case <-gcChan:
9421061 log.Debug("running periodic garbage collection")
9431062 if err := runGC(cfg); err != nil {
9441063 log.Error("periodic gc", "err", err)
......@@ -1560,21 +1679,21 @@ func syncJobArtifacts(cfg *Cfg, repoName, jobID string, log *slog.Logger) error
15601679 }
15611680 sshArgs = fmt.Sprintf("-F ~/.ssh/config -i %s%s", cfg.KeyLocation, certFile)
15621681 }
1563- // Append "/" so rsync copies into the destination directory,
1564- // not as a subdirectory named after the source.
1682+ // Source is jobDir (without trailing slash) so rsync creates
1683+ // the jobID subdirectory inside the destination folder.
15651684 dest := event.ArtifactDest
15661685 if !strings.HasSuffix(dest, "/") {
15671686 dest += "/"
15681687 }
15691688 var cmd *exec.Cmd
15701689 if sshArgs != "" {
1571- cmd = exec.Command("rsync", "-e", sshArgs, "-rv", jobDir+"/", dest)
1690+ cmd = exec.Command("rsync", "-e", sshArgs, "-rv", jobDir, dest)
15721691 } else {
1573- cmd = exec.Command("rsync", "-rv", jobDir+"/", dest)
1692+ cmd = exec.Command("rsync", "-rv", jobDir, dest)
15741693 }
15751694 rsyncCmd := fmt.Sprintf("rsync %s %s %s",
15761695 strings.TrimLeft(cmd.Args[1], "-"),
1577- jobDir+"/", dest)
1696+ jobDir, dest)
15781697 log.Info("rsync", "cmd", rsyncCmd)
15791698 return runCmd(cmd, log)
15801699 }
......@@ -1730,7 +1849,7 @@ func runGC(cfg *Cfg) error {
17301849
17311850 var toKill []string
17321851 for _, s := range sessions {
1733- if !strings.HasPrefix(s.Name, "ci.") {
1852+ if !strings.HasPrefix(s.Name, "ci.") && !strings.HasPrefix(s.Name, "local.") {
17341853 continue
17351854 }
17361855
......@@ -1951,3 +2070,248 @@ func fmtDuration(created, ended string) string {
19512070 }
19522071 return fmt.Sprintf("%.1fs", secs)
19532072 }
2073+
2074+func runLocal(cfg *Cfg, dest string) error {
2075+ cwd, err := os.Getwd()
2076+ if err != nil {
2077+ return fmt.Errorf("get working directory: %w", err)
2078+ }
2079+
2080+ picoPath := filepath.Join(cwd, "pico.sh")
2081+ if _, err := os.Stat(picoPath); os.IsNotExist(err) {
2082+ printMissingPicoHelp(cwd)
2083+ return fmt.Errorf("no pico.sh found in %s", cwd)
2084+ }
2085+
2086+ repoName := filepath.Base(cwd)
2087+ jobID := fmt.Sprintf("local-%d", time.Now().Unix())
2088+
2089+ eventData := &Event{
2090+ Type: "local",
2091+ Name: repoName,
2092+ JobID: jobID,
2093+ Workspace: cwd,
2094+ ArtifactDest: dest,
2095+ }
2096+
2097+ // Process -e / --env flags
2098+ for _, envPair := range cfg.EnvVars {
2099+ parts := strings.SplitN(envPair, "=", 2)
2100+ val := ""
2101+ if len(parts) == 2 {
2102+ val = parts[1]
2103+ }
2104+ key := parts[0]
2105+ switch key {
2106+ case "PICI_REPO":
2107+ eventData.Name = val
2108+ case "PICI_JOB":
2109+ eventData.JobID = val
2110+ case "PICI_EVENT":
2111+ eventData.Type = val
2112+ case "PICI_BRANCH":
2113+ eventData.Branch = val
2114+ case "PICI_COMMIT":
2115+ eventData.Commit = val
2116+ case "PICI_TAG":
2117+ eventData.Tag = val
2118+ }
2119+ _ = os.Setenv(key, val)
2120+ }
2121+
2122+ // Auto-detect git commit SHA and branch if not explicitly provided via -e
2123+ if eventData.Commit == "" {
2124+ eventData.Commit = detectGitCommit(cwd)
2125+ }
2126+ if eventData.Branch == "" {
2127+ eventData.Branch = detectGitBranch(cwd)
2128+ }
2129+
2130+ // Always block and output human format for local runs
2131+ cfg.Wait = true
2132+ cfg.HumanOutput = true
2133+
2134+ logger := cfg.Logger
2135+ if logger == nil {
2136+ logger = newLogger("ci", "info")
2137+ }
2138+ log := logger.With("repo", eventData.Name, "job_id", jobID, "mode", "local")
2139+
2140+ // Set up workspace in temp directory
2141+ wk := &WorkspaceRsync{
2142+ Cfg: cfg,
2143+ Logger: log,
2144+ Source: cwd,
2145+ }
2146+
2147+ eng := &JobEngine{
2148+ Logger: log,
2149+ Cfg: cfg,
2150+ Wk: wk,
2151+ Ev: eventData,
2152+ JobID: jobID,
2153+ }
2154+
2155+ defer func() {
2156+ if err := eng.Cleanup(); err != nil {
2157+ log.Error("cleanup workspace", "err", err)
2158+ }
2159+ }()
2160+
2161+ fmt.Fprintf(os.Stdout, "🚀 starting local job local.%s.%s\n", eventData.Name, jobID) //nolint:errcheck
2162+ fmt.Fprintln(os.Stdout, "📦 syncing workspace to temp directory...") //nolint:errcheck
2163+ if err := eng.Setup(); err != nil {
2164+ return fmt.Errorf("workspace setup: %w", err)
2165+ }
2166+ log.Debug("workspace directory", "dir", eng.Wk.GetDir())
2167+
2168+ // Store event.json in artifact dir
2169+ eventDir := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID)
2170+ artifactsDir := filepath.Join(eventDir, "artifacts")
2171+ if err := os.MkdirAll(artifactsDir, 0755); err != nil {
2172+ log.Error("create artifacts dir", "err", err)
2173+ } else {
2174+ eventBytes, _ := json.Marshal(eventData)
2175+ _ = os.WriteFile(filepath.Join(artifactsDir, "event.json"), eventBytes, 0644)
2176+ }
2177+
2178+ // Write attestation.json
2179+ hostname, _ := os.Hostname()
2180+ attestation := map[string]interface{}{
2181+ "runner": map[string]string{
2182+ "hostname": hostname,
2183+ "os": runtimeOS(),
2184+ "arch": runtimeArch(),
2185+ },
2186+ "provenance": map[string]string{
2187+ "repo": eventData.Name,
2188+ "branch": eventData.Branch,
2189+ "commit": eventData.Commit,
2190+ },
2191+ "workspace_checksum": eng.Wk.Checksum(),
2192+ }
2193+ attestationBytes, _ := json.Marshal(attestation)
2194+ _ = os.WriteFile(filepath.Join(artifactsDir, "attestation.json"), attestationBytes, 0644)
2195+
2196+ manifest, err := eng.FindManifest()
2197+ if err != nil {
2198+ return err
2199+ }
2200+
2201+ fmt.Fprintln(os.Stdout, "🏃 launching sessions...") //nolint:errcheck
2202+ if err := eng.Run(manifest); err != nil {
2203+ return fmt.Errorf("run: %w", err)
2204+ }
2205+
2206+ // Wait for completion & print live progress
2207+ if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
2208+ return fmt.Errorf("wait: %w", err)
2209+ }
2210+
2211+ // Generate and stage full HTML/txt artifacts and index
2212+ if err := stageLocalArtifacts(cfg, log, eventData.Name, jobID); err != nil {
2213+ log.Error("stage local artifacts", "err", err)
2214+ }
2215+
2216+ indexFile := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID, "index.html")
2217+ fmt.Fprintf(os.Stdout, "📊 local html report: file://%s\n", indexFile) //nolint:errcheck
2218+
2219+ // Sync to destination if specified
2220+ if dest != "" {
2221+ targetDest := strings.TrimSuffix(dest, "/") + "/" + jobID
2222+ fmt.Fprintf(os.Stdout, "🔄 rsyncing artifacts to %s...\n", dest) //nolint:errcheck
2223+ if err := syncJobArtifacts(cfg, eventData.Name, jobID, log); err != nil {
2224+ return fmt.Errorf("sync artifacts: %w", err)
2225+ }
2226+ fmt.Fprintf(os.Stdout, "✅ artifacts rsynced to %s\n", targetDest) //nolint:errcheck
2227+ }
2228+
2229+ return nil
2230+}
2231+
2232+func stageLocalArtifacts(cfg *Cfg, log *slog.Logger, repoName, jobID string) error {
2233+ listOutput, err := exec.Command("zmx", "list").CombinedOutput()
2234+ if err != nil {
2235+ return fmt.Errorf("zmx list: %w", err)
2236+ }
2237+ sessions := parseZMXList(string(listOutput))
2238+ var localSessions []SessionInfo
2239+ for _, s := range sessions {
2240+ if strings.HasPrefix(s.Name, "local.") {
2241+ localSessions = append(localSessions, s)
2242+ }
2243+ }
2244+ prefix := fmt.Sprintf("local.%s.%s.", repoName, jobID)
2245+
2246+ var jobSessions []SessionInfo
2247+ for _, s := range localSessions {
2248+ if strings.HasPrefix(s.Name, prefix) {
2249+ s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
2250+ jobSessions = append(jobSessions, s)
2251+ }
2252+ }
2253+
2254+ for _, s := range jobSessions {
2255+ sessionStatus := "running"
2256+ sessionDuration := fmtDuration(s.Created, fmt.Sprintf("%d", time.Now().Unix()))
2257+ sessionExitCode := ""
2258+ if s.Ended != "" {
2259+ sessionDuration = fmtDuration(s.Created, s.Ended)
2260+ if s.ExitCode == "0" {
2261+ sessionStatus = "success"
2262+ sessionExitCode = "0"
2263+ } else {
2264+ sessionStatus = "failed"
2265+ sessionExitCode = s.ExitCode
2266+ }
2267+ }
2268+
2269+ html, err := fetchHistoryHTML(s.Name, repoName, jobID, sessionStatus, sessionDuration, sessionExitCode)
2270+ if err == nil {
2271+ _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, html, ".html")
2272+ }
2273+ plain, err := fetchHistoryPlain(s.Name)
2274+ if err == nil {
2275+ _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, plain, ".txt")
2276+ }
2277+ }
2278+
2279+ // Write published sentinel
2280+ exitCode, status := resolveJobExitCode(jobSessions)
2281+ sentinel := filepath.Join(cfg.ArtifactDir, repoName, jobID, "artifacts", "published.json")
2282+ published := map[string]interface{}{
2283+ "status": status,
2284+ "exit_code": exitCode,
2285+ "job_id": jobID,
2286+ "finished_at": time.Now().UTC().Format(time.RFC3339),
2287+ }
2288+ publishedJSON, _ := json.Marshal(published)
2289+ _ = os.WriteFile(sentinel, publishedJSON, 0644)
2290+
2291+ indexHTML, indexTXT := generateJobIndex(cfg.ArtifactDir, repoName, jobID, jobSessions)
2292+ _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexHTML, ".html")
2293+ _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexTXT, ".txt")
2294+ if styles, err := loadStyles(); err == nil {
2295+ _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "styles", styles, ".css")
2296+ }
2297+
2298+ return nil
2299+}
2300+
2301+func detectGitCommit(dir string) string {
2302+ cmd := exec.Command("git", "-C", dir, "rev-parse", "HEAD")
2303+ out, err := cmd.Output()
2304+ if err != nil {
2305+ return ""
2306+ }
2307+ return strings.TrimSpace(string(out))
2308+}
2309+
2310+func detectGitBranch(dir string) string {
2311+ cmd := exec.Command("git", "-C", dir, "rev-parse", "--abbrev-ref", "HEAD")
2312+ out, err := cmd.Output()
2313+ if err != nil || strings.TrimSpace(string(out)) == "HEAD" {
2314+ return ""
2315+ }
2316+ return strings.TrimSpace(string(out))
2317+}
+104, -4
......@@ -25,17 +25,20 @@ func TestE2E_RunnerWithZMXSessions(t *testing.T) {
2525 if testing.Short() {
2626 t.Skip("skip integration test")
2727 }
28- if _, err := exec.LookPath("zmx"); err != nil {
28+ zmxPath, err := exec.LookPath("zmx")
29+ if err != nil {
2930 t.Skip("zmx not found, skipping integration test")
3031 }
32+ zmxDir := filepath.Dir(zmxPath)
3133
3234 // 1. Create workspace with pico.sh that spawns zmx sessions
3335 workspaceDir := t.TempDir()
34- picoSh := `#!/usr/bin/env bash
36+ picoSh := fmt.Sprintf(`#!/usr/bin/env bash
3537 set -e
38+export PATH="%s:$PATH"
3639 zmx run step1 echo "hello from step1"
3740 zmx run step2 echo "hello from step2"
38-`
41+`, zmxDir)
3942 if err := os.WriteFile(filepath.Join(workspaceDir, "pico.sh"), []byte(picoSh), 0755); err != nil {
4043 t.Fatalf("write pico.sh: %v", err)
4144 }
......@@ -43,9 +46,14 @@ zmx run step2 echo "hello from step2"
4346 // 2. Create config
4447 artifactDir := t.TempDir()
4548 ctx, cancel := context.WithCancel(context.Background())
49+ testJobID := fmt.Sprintf("j-%d", time.Now().UnixNano()%100000000)
50+ t.Cleanup(func() {
51+ _ = exec.Command("zmx", "kill", "-f", fmt.Sprintf("ci.test-repo.%s", testJobID)).Run()
52+ })
4653 event := Event{
4754 Type: "build",
4855 Name: "test-repo",
56+ JobID: testJobID,
4957 Workspace: workspaceDir,
5058 }
5159 eventJSON, _ := json.Marshal(event)
......@@ -109,7 +117,7 @@ zmx run step2 echo "hello from step2"
109117 }
110118
111119 // Ignore statuses from unrelated jobs (e.g., leftover sessions from previous tests)
112- if p.Name != "test-repo" {
120+ if p.Name != "test-repo" || p.JobID != testJobID {
113121 continue
114122 }
115123
......@@ -690,3 +698,95 @@ func TestResolveJobExitCode(t *testing.T) {
690698 })
691699 }
692700 }
701+
702+func TestRunLocal_MissingPico(t *testing.T) {
703+ tempDir := t.TempDir()
704+ origWd, _ := os.Getwd()
705+ defer func() { _ = os.Chdir(origWd) }()
706+ _ = os.Chdir(tempDir)
707+
708+ cfg := &Cfg{
709+ ArtifactDir: t.TempDir(),
710+ }
711+ err := runLocal(cfg, "")
712+ if err == nil {
713+ t.Fatal("expected error when pico.sh is missing")
714+ }
715+}
716+
717+func TestRunLocal_EnvOverrides(t *testing.T) {
718+ tempDir := t.TempDir()
719+ origWd, _ := os.Getwd()
720+ defer func() { _ = os.Chdir(origWd) }()
721+ _ = os.Chdir(tempDir)
722+
723+ picoContent := `#!/usr/bin/env bash
724+echo "hello from pico"
725+`
726+ if err := os.WriteFile("pico.sh", []byte(picoContent), 0755); err != nil {
727+ t.Fatal(err)
728+ }
729+
730+ cfg := &Cfg{
731+ ArtifactDir: t.TempDir(),
732+ EnvVars: envList{"PICI_REPO=custom-repo", "CUSTOM_VAR=hello"},
733+ }
734+
735+ _ = runLocal(cfg, "")
736+
737+ if os.Getenv("CUSTOM_VAR") != "hello" {
738+ t.Errorf("expected CUSTOM_VAR to be hello, got %q", os.Getenv("CUSTOM_VAR"))
739+ }
740+}
741+
742+func TestDetectGitCommit(t *testing.T) {
743+ cwd, err := os.Getwd()
744+ if err != nil {
745+ t.Fatal(err)
746+ }
747+ commit := detectGitCommit(cwd)
748+ if commit == "" {
749+ t.Log("git commit sha not detected (not a git repository or git unavailable)")
750+ } else {
751+ t.Logf("detected git commit sha: %s", commit)
752+ }
753+}
754+
755+func TestWaitAndReport_Cancellation(t *testing.T) {
756+ ctx, cancel := context.WithCancel(context.Background())
757+ cfg := &Cfg{
758+ Ctx: ctx,
759+ Cancel: cancel,
760+ MonitorInterval: 100 * time.Millisecond,
761+ }
762+
763+ jobID := fmt.Sprintf("canceltest-%d", time.Now().UnixNano())
764+ prefix := "local.testrepo." + jobID + "."
765+ runnerSession := prefix + "runner"
766+
767+ cmd := exec.Command("zmx", "run", runnerSession, "-d", "sleep", "30")
768+ if err := cmd.Run(); err != nil {
769+ t.Fatalf("failed to start zmx session: %v", err)
770+ }
771+
772+ go func() {
773+ time.Sleep(200 * time.Millisecond)
774+ cancel()
775+ }()
776+
777+ err := waitAndReport(cfg, nil, "testrepo", jobID, "local")
778+ if err == nil {
779+ t.Error("expected error when waitAndReport is cancelled, got nil")
780+ }
781+
782+ time.Sleep(200 * time.Millisecond)
783+ listOutput, _ := exec.Command("zmx", "list").CombinedOutput()
784+ sessions := parseZMXList(string(listOutput))
785+ for _, s := range sessions {
786+ if s.Name == runnerSession {
787+ if s.Ended == "" {
788+ t.Errorf("session %s should have been killed on cancellation", runnerSession)
789+ }
790+ }
791+ }
792+}