main
main.go
Eric Bower
·
2026-08-16
1package main
2
3import (
4 "archive/tar"
5 "bufio"
6 "context"
7 "crypto/sha256"
8 "embed"
9 "encoding/json"
10 "errors"
11 "flag"
12 "fmt"
13 "html/template"
14 "io"
15 "io/fs"
16 "log/slog"
17 "os"
18 "os/exec"
19 "path/filepath"
20 "runtime"
21 "sort"
22 "strconv"
23 "strings"
24 "time"
25)
26
27//go:embed tmpl/*
28var tmplFS embed.FS
29
30type WorkspaceFactory func(cfg *Cfg, logger *slog.Logger, source string) Workspace
31
32func defaultWorkspaceFactory(cfg *Cfg, logger *slog.Logger, source string) Workspace {
33 if cfg != nil && cfg.InPlace {
34 return &WorkspaceInPlace{
35 Cfg: cfg,
36 Logger: logger,
37 Source: source,
38 }
39 }
40 if strings.HasSuffix(source, ".tar") {
41 return &WorkspaceTar{
42 Cfg: cfg,
43 Logger: logger,
44 Source: source,
45 }
46 }
47 return &WorkspaceRsync{
48 Cfg: cfg,
49 Logger: logger,
50 Source: source,
51 }
52}
53
54type envList []string
55
56func (e *envList) String() string {
57 return strings.Join(*e, ", ")
58}
59
60func (e *envList) Set(value string) error {
61 *e = append(*e, value)
62 return nil
63}
64
65type Cfg struct {
66 Logger *slog.Logger
67 Ctx context.Context
68 Cancel context.CancelFunc
69 KeyLocation string
70 CertificateLocation string
71 ArtifactDir string
72 Event string // event JSON passed via --event flag
73 EventSource io.ReadCloser // when set, used directly as the event source (for testing)
74 MonitorInterval time.Duration
75 GCInterval time.Duration
76 NewWorkspace WorkspaceFactory
77 StatusOutput io.Writer // where status JSONL is written (default: os.Stdout)
78 IncludeRunning bool // emit running status updates in addition to terminal
79 HumanOutput bool // human-readable output instead of JSONL / slog
80 Wait bool // block until job completes, print history and summary
81 InPlace bool // execute workspace in-place without rsyncing to /tmp
82 SummaryFile string // write markdown summary to this file path
83 DebugOnFail bool // open interactive pipe.pico.sh debug bridge on failure
84 DebugTimeout time.Duration // idle timeout for debug bridge (default: 15m)
85 EnvVars envList // custom environment variables passed via -e / -env
86 SessionPrefix string // session prefix filter for monitor (default: "ci.")
87}
88
89// JobFailedError indicates that one or more sessions within a job failed,
90// carrying the resolved non-zero exit code.
91type JobFailedError struct {
92 ExitCode int
93}
94
95func (e *JobFailedError) Error() string {
96 return fmt.Sprintf("job failed with exit code %d", e.ExitCode)
97}
98
99type Event struct {
100 Type string `json:"type"`
101 Name string `json:"name"`
102 JobID string `json:"job_id,omitempty"` // optional producer-set ID
103 Workspace string `json:"workspace"`
104 Branch string `json:"branch"`
105 Tag string `json:"tag"`
106 Commit string `json:"commit"`
107 ArtifactDest string `json:"artifact_dest"`
108 ForcePush bool `json:"force_push,omitempty"`
109}
110
111func NewCfg() (*Cfg, string, bool) {
112 var keyLoc, certLoc, artifactDir, event string
113 var monitorInterval time.Duration
114 var gcInterval time.Duration
115 var logLevel string
116 var envVars envList
117 var sessionPrefix string
118 var debugTimeout time.Duration
119 flag.StringVar(&keyLoc, "pk", "", "ssh private key used to authenticate with pico services")
120 flag.StringVar(&certLoc, "ck", "", "ssh certificate public key used to authenticate with pico services (only required if using ssh certificates)")
121 flag.StringVar(&artifactDir, "artifact-dir", "/tmp/pici-artifacts", "local directory to stage artifacts")
122 flag.StringVar(&event, "event", "", "event JSON to run (alternative to reading from stdin)")
123 flag.StringVar(&sessionPrefix, "prefix", "ci.", "session prefix filter for monitor (default: ci.)")
124 flag.DurationVar(&monitorInterval, "monitor-interval", 5*time.Second, "interval for monitoring zmx sessions")
125 flag.DurationVar(&gcInterval, "gc-interval", 10*time.Minute, "interval for garbage collection in monitor (0 to disable)")
126 flag.StringVar(&logLevel, "log-level", "info", "log level: debug, info, warn, error")
127 flag.Var(&envVars, "e", "environment variable in KEY=VAL format (can be specified multiple times)")
128 flag.Var(&envVars, "env", "environment variable in KEY=VAL format (can be specified multiple times)")
129 var includeRunning bool
130 var human bool
131 var wait bool
132 var inPlace bool
133 var summaryFile string
134 var debugOnFail bool
135 flag.BoolVar(&includeRunning, "include-running", false, "emit running status updates in addition to terminal (default: terminal only)")
136 flag.BoolVar(&human, "human", false, "human-readable output (default: JSONL / slog)")
137 flag.BoolVar(&wait, "wait", false, "block until job completes, printing session history and summary")
138 flag.BoolVar(&inPlace, "in-place", false, "execute workspace in-place without rsyncing to /tmp")
139 flag.StringVar(&summaryFile, "summary-file", "", "file path to write markdown job summary")
140 flag.BoolVar(&debugOnFail, "debug-on-fail", false, "open interactive pipe.pico.sh debug bridge on failure")
141 flag.DurationVar(&debugTimeout, "debug-timeout", 15*time.Minute, "idle timeout for interactive debug bridge")
142
143 // Split args so the subcommand (first non-flag arg) doesn't block
144 // flags that appear after it: "pici runner --wait" works.
145 flags, cmd, wantHelp := splitCommand(os.Args[1:])
146 if err := flag.CommandLine.Parse(flags); err != nil {
147 fmt.Fprintf(os.Stderr, "failed to parse flags: %v\n", err)
148 os.Exit(1)
149 }
150
151 for _, envPair := range envVars {
152 parts := strings.SplitN(envPair, "=", 2)
153 if len(parts) == 2 && parts[0] == "PICI_DEBUG" && (parts[1] == "1" || parts[1] == "true") {
154 debugOnFail = true
155 }
156 }
157 if envVal := os.Getenv("PICI_DEBUG"); envVal == "1" || envVal == "true" {
158 debugOnFail = true
159 }
160
161 logger := newLogger("ci", logLevel)
162 ctx, cancel := context.WithCancel(context.Background())
163 return &Cfg{
164 NewWorkspace: defaultWorkspaceFactory,
165 Logger: logger.With("key_loc", keyLoc, "cert_loc", certLoc),
166 Ctx: ctx,
167 Cancel: cancel,
168 KeyLocation: keyLoc,
169 CertificateLocation: certLoc,
170 ArtifactDir: artifactDir,
171 Event: event,
172 MonitorInterval: monitorInterval,
173 GCInterval: gcInterval,
174 IncludeRunning: includeRunning,
175 HumanOutput: human,
176 Wait: wait,
177 InPlace: inPlace,
178 SummaryFile: summaryFile,
179 DebugOnFail: debugOnFail,
180 DebugTimeout: debugTimeout,
181 EnvVars: envVars,
182 SessionPrefix: sessionPrefix,
183 }, cmd, wantHelp
184}
185
186func isKnownSubcommand(s string) bool {
187 switch s {
188 case "runner", "monitor", "cancel", "gc", "status", "orca", "run", "debug", "help":
189 return true
190 default:
191 return false
192 }
193}
194
195// splitCommand separates the first non-flag argument (the subcommand) from
196// the rest of the flags, so "runner --wait" becomes flags=["--wait"], cmd="runner".
197// It strips --help/help so we can print custom help per subcommand.
198func splitCommand(args []string) (flags []string, cmd string, wantHelp bool) {
199 flags = make([]string, 0, len(args))
200 for _, arg := range args {
201 if arg == "--help" || arg == "help" || arg == "-h" {
202 wantHelp = true
203 continue
204 }
205 if cmd == "" && isKnownSubcommand(arg) {
206 cmd = arg
207 } else {
208 flags = append(flags, arg)
209 }
210 }
211 return flags, cmd, wantHelp
212}
213
214func parseLogLevel(s string) slog.Level {
215 switch strings.ToLower(s) {
216 case "debug":
217 return slog.LevelDebug
218 case "warn":
219 return slog.LevelWarn
220 case "error":
221 return slog.LevelError
222 default:
223 return slog.LevelInfo
224 }
225}
226
227func newLogger(space string, levelStr string) *slog.Logger {
228 lvl := parseLogLevel(levelStr)
229 return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{
230 Level: lvl,
231 })).With("service", space)
232}
233
234func printMainHelp() {
235 fmt.Println(`pici - minimal parallel CI runner & monitor powered by zmx
236
237HOW IT WORKS: pici runs your project's CI pipeline defined in pico.sh:
238 1. Syncs workspace to a clean temporary directory (/tmp).
239 2. Executes pico.sh inside zmx, running each task in its own isolated PTY.
240 3. Bring your own isolation: tasks run on host unless wrapped (e.g. Docker, Podman, Nix).
241 4. Tasks execute serially ('zmx run <name> <cmd>') or concurrently in detached mode ('zmx run <name> -d <cmd>').
242 5. Concurrent background tasks are joined using 'zmx wait "*"' to block until all complete.
243 6. Streams live terminal output & generates HTML reports in /tmp/pici-artifacts/.
244 7. Interactive debugging: on failure, attach with 'zmx attach <session>' to inspect logs or rerun commands.
245 8. Optionally rsyncs rendered logs & artifacts to a destination when specified.
246
247PROJECT SETUP
248 1. Create a pico.sh script in your project root:
249
250 #!/usr/bin/env bash
251 set -euo pipefail
252
253 # Read environment metadata (with fallbacks for standalone execution)
254 JOB_ID="${PICI_JOB:-local}"
255 REPO="${PICI_REPO:-myrepo}"
256 BRANCH="${PICI_BRANCH:-main}"
257 COMMIT="${PICI_COMMIT:-dev}"
258 EVENT="${PICI_EVENT:-local}"
259 ZMX_SESSION_PREFIX="${ZMX_SESSION_PREFIX:-local.}"
260
261 echo "Running CI for $REPO ($BRANCH@$COMMIT, job: $JOB_ID, event: $EVENT)"
262
263 # Run serial setup step on host or container
264 zmx run setup npm install
265
266 # Run concurrent steps using Docker for environment isolation
267 zmx run lint -d docker run --rm -v "$(pwd):/app" -w /app golangci/golangci-lint golangci-lint run
268 zmx run test -d docker run --rm -v "$(pwd):/app" -w /app golang:1.26 go test ./...
269
270 # Wait for all background steps to finish
271 zmx wait "*"
272
273 2. Make it executable: chmod +x pico.sh
274 3. Run locally: pici
275
276ENVIRONMENT VARIABLES: pici injects these vars into pico.sh:
277 PICI_JOB Unique job identifier (timestamp or commit SHA)
278 PICI_REPO Repository name (default: directory basename)
279 PICI_EVENT Event type ('local', 'git.push', 'git.tag')
280 PICI_BRANCH Git branch name (auto-detected)
281 PICI_COMMIT Git commit SHA (auto-detected)
282 PICI_TAG Git tag name (set for tag events)
283 PICI_ARTIFACTS_DIR Path to job artifacts directory for staging assets
284 ZMX_SESSION_PREFIX Internal session prefix managed by pici
285
286 Pass custom env vars or overrides with -e / -env (e.g. pici -e CGO_ENABLED=0).
287
288LOCAL DEVELOPER USAGE
289 pici [destination] [flags] Run ./pico.sh locally in /tmp & render HTML logs
290 pici pgs.sh:/my-site Run locally and rsync HTML logs to destination
291 pici run [destination] Explicit alias for local run
292 pici debug <repo/job_id | session> Attach to active zmx session for debugging
293
294DAEMON & SERVICE COMMANDS
295 pici runner Execute CI job from event JSON payload (stdin/flag)
296 pici monitor Poll ci.* zmx sessions & stage/sync HTML artifacts
297 pici cancel Cancel active running jobs for a repository
298 pici gc Clean up stale/finished zmx sessions & artifacts
299
300SUBCOMMAND HELP
301 pici <command> --help Show detailed help for a specific command (e.g. pici runner --help)
302
303FLAGS
304 -e, -env <KEY=VAL> Set or override environment variable for pico.sh
305 -in-place Execute workspace in-place without copying to /tmp
306 -summary-file <path> Write markdown job summary table to file
307 -debug-on-fail Open interactive pipe.pico.sh debug bridge on failure
308 -debug-timeout <dur> Idle timeout for interactive debug bridge (default: 15m)
309 -pk <path> SSH private key
310 -ck <path> SSH public certificate key
311 -artifact-dir <path> Artifact staging directory (default: /tmp/pici-artifacts)
312 -log-level <level> Log level: debug, info, warn, error`)
313}
314
315func printMissingPicoHelp(cwd string) {
316 fmt.Printf(`❌ Error: no pico.sh found in %s
317
318To run local CI tasks, create a pico.sh script in your project root:
319
320 #!/usr/bin/env bash
321 set -euo pipefail
322
323 # Read environment metadata (with fallbacks for standalone execution)
324 JOB_ID="${PICI_JOB:-local}"
325 REPO="${PICI_REPO:-myrepo}"
326 BRANCH="${PICI_BRANCH:-main}"
327 COMMIT="${PICI_COMMIT:-dev}"
328 EVENT="${PICI_EVENT:-local}"
329 ZMX_SESSION_PREFIX="${ZMX_SESSION_PREFIX:-local.}"
330
331 echo "Running CI for $REPO ($BRANCH@$COMMIT, job: $JOB_ID, event: $EVENT)"
332
333 # Run serial setup step on host or container
334 zmx run setup npm install
335
336 # Run concurrent steps using Docker for environment isolation
337 zmx run lint -d docker run --rm -v "$(pwd):/app" -w /app golangci/golangci-lint golangci-lint run
338 zmx run test -d docker run --rm -v "$(pwd):/app" -w /app golang:1.26 go test ./...
339
340 # Wait for all background steps to finish
341 zmx wait "*"
342
343For full command documentation and daemon options, run:
344 pici help
345
346`, cwd)
347}
348
349func main() {
350 cfg, cmd, wantHelp := NewCfg()
351
352 cfg.Logger.Debug("setting up ci", "cfg", cfg)
353 cfg.Logger.Debug("running cmd", "cmd", cmd)
354
355 if wantHelp {
356 switch cmd {
357 case "runner":
358 printRunnerHelp()
359 case "monitor":
360 printMonitorHelp()
361 default:
362 printMainHelp()
363 }
364 return
365 }
366
367 switch cmd {
368 case "runner":
369 cfg.Logger.Debug("starting runner")
370 if err := RunRunner(cfg); err != nil {
371 var failedErr *JobFailedError
372 if errors.As(err, &failedErr) {
373 cfg.Logger.Error("runner failed", "err", err, "exit_code", failedErr.ExitCode)
374 os.Exit(failedErr.ExitCode)
375 }
376 cfg.Logger.Error("runner failed", "err", err)
377 os.Exit(1)
378 }
379 case "cancel":
380 cfg.Logger.Debug("starting cancel handler")
381 if err := runCancel(cfg); err != nil {
382 cfg.Logger.Error("cancel failed", "err", err)
383 os.Exit(1)
384 }
385 case "gc":
386 cfg.Logger.Debug("starting garbage collection")
387 if err := runGC(cfg); err != nil {
388 cfg.Logger.Error("gc failed", "err", err)
389 os.Exit(1)
390 }
391 case "monitor":
392 cfg.Logger.Debug("starting monitor")
393 if err := runMonitor(cfg); err != nil {
394 cfg.Logger.Error("monitor failed", "err", err)
395 os.Exit(1)
396 }
397 case "status":
398 cfg.Logger.Debug("starting status updater")
399 case "debug":
400 cfg.Logger.Debug("starting debug session")
401 target := ""
402 if flag.NArg() > 0 {
403 target = flag.Arg(0)
404 }
405 if err := runDebug(cfg, target); err != nil {
406 cfg.Logger.Error("debug failed", "err", err)
407 os.Exit(1)
408 }
409 case "help":
410 printMainHelp()
411 case "run", "":
412 dest := ""
413 if flag.NArg() > 0 {
414 dest = flag.Arg(0)
415 }
416 if err := runLocal(cfg, dest); err != nil {
417 var failedErr *JobFailedError
418 if errors.As(err, &failedErr) {
419 cfg.Logger.Error("local run failed", "err", err, "exit_code", failedErr.ExitCode)
420 os.Exit(failedErr.ExitCode)
421 }
422 cfg.Logger.Error("local run failed", "err", err)
423 os.Exit(1)
424 }
425 default:
426 dest := cmd
427 if err := runLocal(cfg, dest); err != nil {
428 var failedErr *JobFailedError
429 if errors.As(err, &failedErr) {
430 cfg.Logger.Error("local run failed", "err", err, "exit_code", failedErr.ExitCode)
431 os.Exit(failedErr.ExitCode)
432 }
433 cfg.Logger.Error("local run failed", "err", err)
434 os.Exit(1)
435 }
436 }
437}
438
439func printMonitorHelp() {
440 fmt.Println(`pici monitor — poll ci.* zmx sessions, stage artifacts, publish status.
441
442USAGE
443 pici monitor [flags] # JSONL to stdout (pipe to webhooks)
444 pici monitor --human # human-readable terminal output
445
446OUTPUT MODES
447 Default (JSONL): One JSON object per line, suitable for piping to notifications.
448 pici monitor > status.jsonl
449 pici monitor | ssh pipe.pico.sh "pub build.status -b=false"
450 pici monitor | while read -r line; do curl -sd"$line" $WEBHOOK; done
451
452 --human: Selfci-style progress output for terminal viewing (do not pipe).
453 pici monitor --human
454 pici monitor --human --include-running
455 [2/3] 🚀 running: myrepo (1m23s)
456 [3/3] ✅ success: myrepo (2m34s)
457
458FLAGS
459 -pk <path> SSH private key for authenticating with pico services
460 -ck <path> SSH certificate public key
461 -artifact-dir <path> Local directory to stage artifacts (default: /tmp/pici-artifacts)
462 -monitor-interval <dur> Poll interval (default: 5s)
463 -gc-interval <dur> Garbage collection interval (default: 10m, 0 to disable)
464 -include-running Emit running status updates in addition to terminal (default: terminal only)
465 -human Human-readable output instead of JSONL (for terminal, not piping)
466 -log-level <level> Log level: debug, info, warn, error (default: info)`)
467}
468
469func printRunnerHelp() {
470 fmt.Println(`pici runner — execute a CI job from an event JSON payload.
471
472USAGE
473 echo '<event-json>' | pici runner [flags]
474 pici runner --event '<event-json>' [flags]
475 echo '<event-json>' | pici runner --wait [flags] # block until done
476
477EVENT JSON FIELDS
478 type (required) Event type, e.g. "push", "merge_request"
479 name (required) Repository name, used for session naming
480 workspace (required) SSH source path to rsync, e.g. "git@github.com:user/repo.git"
481 artifact_dest (optional) SSH destination for artifact sync, e.g. "user@host:/path"
482
483
484EXAMPLE
485 echo '{"type":"push","name":"myrepo","workspace":"git@github.com:user/myrepo.git"}' | pici runner
486
487ENV VARS IN pico.sh
488 The runner exports these environment variables for your pico.sh script:
489 PICI_JOB Unique job identifier (e.g. "a3f2b8c1")
490 PICI_REPO Repository name (from event "name" field)
491 PICI_EVENT Event type (e.g. "git.push", "git.tag")
492 PICI_TAG Git tag name (set only for git.tag events)
493 PICI_BRANCH Git branch name (set only for git.push events)
494 PICI_ARTIFACTS_DIR Path to job artifacts directory for staging assets
495
496 Note: pico.sh runs with the workspace as its working directory, so
497 $(pwd) gives you the workspace path directly.
498
499 Use them with defaults so pico.sh works standalone:
500 JOB="${PICI_JOB:-local}"
501 REPO="${PICI_REPO:-unknown}"
502
503FLAGS
504 -pk <path> SSH private key for authenticating with pico services
505 -ck <path> SSH certificate public key (required when using SSH certificates)
506 -event <json> Event JSON string (alternative to reading from stdin)
507 -artifact-dir <path> Local directory to stage artifacts (default: /tmp/pici-artifacts)
508 -log-level <level> Log level: debug, info, warn, error (default: info)
509 -human Human-readable output (default: enabled for runner)
510 -wait Block until job completes, printing session history and summary`)
511
512}
513
514func RunRunner(cfg *Cfg) error {
515 var payload string
516 if cfg.EventSource != nil {
517 data, err := io.ReadAll(cfg.EventSource)
518 if err != nil {
519 return fmt.Errorf("read event source: %w", err)
520 }
521 payload = strings.TrimSpace(string(data))
522 } else if cfg.Event != "" {
523 payload = cfg.Event
524 } else {
525 data, err := io.ReadAll(os.Stdin)
526 if err != nil {
527 return fmt.Errorf("read stdin: %w", err)
528 }
529 payload = strings.TrimSpace(string(data))
530 }
531
532 var eventData Event
533 if err := json.Unmarshal([]byte(payload), &eventData); err != nil {
534 return fmt.Errorf("unmarshal event: %w", err)
535 }
536
537 // Validate required fields
538 if eventData.Type == "" {
539 return fmt.Errorf("event missing required field: type")
540 }
541 if eventData.Name == "" {
542 return fmt.Errorf("event missing required field: name")
543 }
544 if eventData.Workspace == "" {
545 return fmt.Errorf("event missing required field: workspace")
546 }
547
548 return eventHandler(cfg, &eventData)
549}
550
551type Workspace interface {
552 Setup() error
553 Cleanup() error
554 GetDir() string
555 // Checksum returns a content hash of the workspace (e.g. sha256 of the tarball).
556 // Returns empty string if not available.
557 Checksum() string
558}
559
560type WorkspaceRsync struct {
561 Cfg *Cfg
562 Logger *slog.Logger
563 Source string
564 Dest string
565 checksum string
566}
567
568func (w *WorkspaceRsync) Setup() error {
569 tempDir, err := os.MkdirTemp("", "pici-*")
570 if err != nil {
571 return err
572 }
573 w.Dest = tempDir
574
575 log := w.Logger.With("source", w.Source, "dest", w.Dest)
576 log.Debug("syncing workspace via rsync")
577
578 var cmd *exec.Cmd
579 if w.Cfg.KeyLocation != "" {
580 sshcmd := fmt.Sprintf(
581 "-F ~/.ssh/config -i %s -o IdentitiesOnly=yes -o CertificateFile %s",
582 w.Cfg.KeyLocation,
583 w.Cfg.CertificateLocation,
584 )
585 cmd = exec.Command("rsync", "-e", sshcmd, "-rv", "--exclude=/.git", "--exclude=/.jj", "--filter=:- .gitignore", w.Source+"/", w.Dest+"/")
586 } else {
587 cmd = exec.Command("rsync", "-rv", "--exclude=/.git", "--exclude=/.jj", "--filter=:- .gitignore", w.Source+"/", w.Dest+"/")
588 }
589 if err := runCmd(cmd, log); err != nil {
590 return err
591 }
592
593 cs, err := computeDirChecksum(w.Dest)
594 if err != nil {
595 log.Error("compute workspace checksum", "err", err)
596 } else {
597 w.checksum = cs
598 log.Debug("workspace checksum", "checksum", w.checksum)
599 }
600 return nil
601}
602
603func (w *WorkspaceRsync) Cleanup() error {
604 if w.Dest != "" {
605 return os.RemoveAll(w.Dest)
606 }
607 return nil
608}
609
610func (w *WorkspaceRsync) GetDir() string {
611 return w.Dest
612}
613
614func (w *WorkspaceRsync) Checksum() string {
615 return w.checksum
616}
617
618type WorkspaceInPlace struct {
619 Cfg *Cfg
620 Logger *slog.Logger
621 Source string
622 checksum string
623}
624
625func (w *WorkspaceInPlace) Setup() error {
626 log := w.Logger.With("source", w.Source)
627 log.Debug("using in-place workspace")
628 cs, err := computeDirChecksum(w.Source)
629 if err != nil {
630 log.Error("compute workspace checksum", "err", err)
631 } else {
632 w.checksum = cs
633 log.Debug("workspace checksum", "checksum", w.checksum)
634 }
635 return nil
636}
637
638func (w *WorkspaceInPlace) Cleanup() error {
639 // In-place workspace is not cleaned up
640 return nil
641}
642
643func (w *WorkspaceInPlace) GetDir() string {
644 return w.Source
645}
646
647func (w *WorkspaceInPlace) Checksum() string {
648 return w.checksum
649}
650
651func computeDirChecksum(dir string) (string, error) {
652 hasher := sha256.New()
653 err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
654 if err != nil {
655 return err
656 }
657 if d.IsDir() {
658 return nil
659 }
660 rel, err := filepath.Rel(dir, path)
661 if err != nil {
662 return err
663 }
664 hasher.Write([]byte(rel))
665 hasher.Write([]byte{0})
666
667 f, err := os.Open(path)
668 if err != nil {
669 return err
670 }
671 _, copyErr := io.Copy(hasher, f)
672 _ = f.Close()
673 if copyErr != nil {
674 return copyErr
675 }
676 return nil
677 })
678 if err != nil {
679 return "", err
680 }
681 return fmt.Sprintf("sha256:%x", hasher.Sum(nil)), nil
682}
683
684type WorkspaceTar struct {
685 Cfg *Cfg
686 Logger *slog.Logger
687 Source string // e.g. "pgs.sh:/private-ci/workspaces/repo_abc123.tar"
688 Dest string
689 checksum string
690}
691
692func (w *WorkspaceTar) Setup() error {
693 tempDir, err := os.MkdirTemp("", "pici-*")
694 if err != nil {
695 return err
696 }
697 w.Dest = tempDir
698
699 log := w.Logger.With("source", w.Source, "dest", w.Dest)
700 log.Debug("downloading and extracting workspace tar")
701
702 // Parse host:path from source
703 host, path := splitSSHSource(w.Source)
704
705 // Use rsync to download the tar file
706 tarPath := filepath.Join(tempDir, "workspace.tar")
707 rsyncCmd := exec.Command("rsync", "-e", "ssh", host+":"+path, tarPath)
708 rsyncCmd.Stderr = os.Stderr
709 if err := rsyncCmd.Run(); err != nil {
710 return fmt.Errorf("rsync download: %w", err)
711 }
712
713 // Compute sha256 checksum of the tarball
714 tarData, err := os.ReadFile(tarPath)
715 if err != nil {
716 return fmt.Errorf("read tar for checksum: %w", err)
717 }
718 w.checksum = fmt.Sprintf("sha256:%x", sha256.Sum256(tarData))
719 log.Debug("workspace checksum", "checksum", w.checksum)
720
721 // Open the tar file for extraction
722 f, err := os.Open(tarPath)
723 if err != nil {
724 return fmt.Errorf("open tar: %w", err)
725 }
726 defer func() {
727 if err := f.Close(); err != nil {
728 log.Error("close tar", "err", err)
729 }
730 }()
731
732 // Extract tar to temp dir
733 tr := tar.NewReader(f)
734 for {
735 hdr, err := tr.Next()
736 if err == io.EOF {
737 break
738 }
739 if err != nil {
740 return fmt.Errorf("tar read: %w", err)
741 }
742
743 target := filepath.Join(tempDir, hdr.Name)
744 switch hdr.Typeflag {
745 case tar.TypeDir:
746 if err := os.MkdirAll(target, 0755); err != nil {
747 return fmt.Errorf("mkdir %s: %w", target, err)
748 }
749 case tar.TypeReg:
750 // Ensure parent dir exists
751 if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
752 return fmt.Errorf("mkdir parent %s: %w", filepath.Dir(target), err)
753 }
754 tf, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY, os.FileMode(hdr.Mode))
755 if err != nil {
756 return fmt.Errorf("open %s: %w", target, err)
757 }
758 if _, err := io.Copy(tf, tr); err != nil {
759 if closeErr := tf.Close(); closeErr != nil {
760 return fmt.Errorf("close %s: %w", target, closeErr)
761 }
762 return fmt.Errorf("write %s: %w", target, err)
763 }
764 if err := tf.Close(); err != nil {
765 return fmt.Errorf("close %s: %w", target, err)
766 }
767 }
768 }
769
770 log.Debug("workspace extracted")
771 return nil
772}
773
774func (w *WorkspaceTar) Checksum() string {
775 return w.checksum
776}
777
778func (w *WorkspaceTar) Cleanup() error {
779 return nil
780}
781
782func (w *WorkspaceTar) GetDir() string {
783 return w.Dest
784}
785
786// splitSSHSource splits "host:path" into host and path.
787func splitSSHSource(source string) (string, string) {
788 idx := strings.Index(source, ":")
789 if idx == -1 {
790 return source, ""
791 }
792 return source[:idx], source[idx+1:]
793}
794
795type JobEngine struct {
796 Wk Workspace
797 Logger *slog.Logger
798 Cfg *Cfg
799 Ev *Event
800 JobID string
801}
802
803type SessionInfo struct {
804 Name string `json:"name"`
805 Short string `json:"short"`
806 PID string `json:"pid"`
807 Clients string `json:"clients"`
808 Created string `json:"created"`
809 StartDir string `json:"start_dir"`
810 Ended string `json:"ended"`
811 ExitCode string `json:"exit_code"`
812}
813
814type StatusPayload struct {
815 Timestamp string `json:"timestamp"`
816 Name string `json:"name"`
817 JobID string `json:"job_id"`
818 Status string `json:"status"`
819 ExitCode *int `json:"exit_code"`
820 Duration string `json:"duration,omitempty"`
821 StartedAt string `json:"started_at,omitempty"`
822 EndedAt string `json:"ended_at,omitempty"`
823 SessionCount int `json:"session_count"`
824 Sessions []SessionInfo `json:"sessions"`
825}
826
827func (eng *JobEngine) Setup() error {
828 return eng.Wk.Setup()
829}
830
831// getDomain maps an event type (e.g. "git.push", "git.tag", "local") to a session domain prefix ("ci", "local", etc.).
832func getDomain(eventType string) string {
833 if eventType == "local" {
834 return "local"
835 }
836 return "ci"
837}
838
839func (eng *JobEngine) Run(manifest string) error {
840 domain := "ci"
841 if eng.Ev != nil {
842 domain = getDomain(eng.Ev.Type)
843 }
844 prefix := fmt.Sprintf("%s.%s.%s.", domain, eng.Ev.Name, eng.JobID)
845 // Child sessions use ".step." sub-prefix so zmx wait "*" inside pico.sh
846 // matches <domain>.<name>.<jobID>.step.* but NOT <domain>.<name>.<jobID>.runner.
847 // This avoids a deadlock where the runner waits for itself.
848 childPrefix := prefix + "step."
849
850 log := eng.Logger.With("manifest", manifest, "prefix", prefix)
851 log.Debug("starting runner session", "session", prefix+"runner")
852
853 runnerName := prefix + "runner"
854 // Name the runner explicitly. Do NOT set ZMX_SESSION_PREFIX for this
855 // outer zmx run call — it would be prepended to the runner's own name.
856 // The prefix is only for child sessions spawned inside pico.sh (via the
857 // exported env var).
858 artifactsDir := filepath.Join(eng.Cfg.ArtifactDir, eng.Ev.Name, eng.JobID, "artifacts")
859 if err := os.MkdirAll(artifactsDir, 0755); err != nil {
860 log.Error("create artifacts dir", "err", err)
861 }
862
863 cmdEnv := make([]string, 0, len(os.Environ())+7)
864 for _, e := range os.Environ() {
865 if !strings.HasPrefix(e, "ZMX_SESSION_PREFIX=") {
866 cmdEnv = append(cmdEnv, e)
867 }
868 }
869 cmdEnv = append(cmdEnv,
870 fmt.Sprintf("PICI_JOB=%s", eng.JobID),
871 fmt.Sprintf("PICI_REPO=%s", eng.Ev.Name),
872 fmt.Sprintf("PICI_EVENT=%s", eng.Ev.Type),
873 fmt.Sprintf("PICI_BRANCH=%s", eng.Ev.Branch),
874 fmt.Sprintf("PICI_ARTIFACTS_DIR=%s", artifactsDir),
875 )
876 if eng.Ev.Tag != "" {
877 cmdEnv = append(cmdEnv, fmt.Sprintf("PICI_TAG=%s", eng.Ev.Tag))
878 }
879 bashCmd := fmt.Sprintf("export ZMX_SESSION_PREFIX=%q; exec bash %q", childPrefix, manifest)
880 cmd := exec.Command("zmx", "run", runnerName, "-d", "bash", "-c", bashCmd)
881 cmd.Env = cmdEnv
882 cmd.Dir = eng.Wk.GetDir()
883
884 if err := cmd.Run(); err != nil {
885 return fmt.Errorf("start runner session: %w", err)
886 }
887
888 return nil
889}
890
891func (eng *JobEngine) Cleanup() error {
892 return eng.Wk.Cleanup()
893}
894
895func (eng *JobEngine) FindManifest() (string, error) {
896 fnames := []string{"pico.sh"}
897 for _, manifest := range fnames {
898 path := filepath.Join(eng.Wk.GetDir(), manifest)
899 if _, err := os.Stat(path); err == nil {
900 return path, nil
901 }
902 }
903 return "", fmt.Errorf("no pico.sh found in %s", eng.Wk.GetDir())
904}
905
906func eventHandler(cfg *Cfg, eventData *Event) error {
907 log := cfg.Logger.With("repo", eventData.Name, "type", eventData.Type)
908
909 jobID := resolveJobID(eventData.Name, eventData.Workspace, eventData.JobID)
910 log = log.With("job_id", jobID)
911
912 // Cancel any existing job for this repo before starting a new one
913 cancelRunningJobs(cfg, log, eventData.Name)
914
915 // Clean up any artifacts from a previous run with the same job ID
916 // (e.g. duplicate event, re-trigger of the same commit).
917 eventDir := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID)
918 if _, err := os.Stat(eventDir); err == nil {
919 if err := os.RemoveAll(eventDir); err != nil {
920 log.Warn("failed to clean old job directory", "err", err, "dir", eventDir)
921 }
922 }
923
924 eventBytes, _ := json.Marshal(eventData)
925 domain := getDomain(eventData.Type)
926 fmt.Fprintf(os.Stdout, "🚀 starting job %s.%s.%s\n", domain, eventData.Name, jobID) //nolint:errcheck
927 fmt.Fprintf(os.Stdout, " event: type=%s name=%s workspace=%s\n", eventData.Type, eventData.Name, eventData.Workspace) //nolint:errcheck
928 newWk := cfg.NewWorkspace
929 if newWk == nil {
930 newWk = defaultWorkspaceFactory
931 }
932 wk := newWk(cfg, log, eventData.Workspace)
933 eng := &JobEngine{
934 Logger: log,
935 Cfg: cfg,
936 Wk: wk,
937 Ev: eventData,
938 JobID: jobID,
939 }
940 var runErr error
941 defer func() {
942 if runErr != nil || cfg.Wait {
943 if err := eng.Cleanup(); err != nil {
944 cfg.Logger.Error("engine cleanup", "err", err)
945 }
946 }
947 }()
948
949 fmt.Fprintf(os.Stdout, "📦 syncing workspace %s\n", eventData.Workspace) //nolint:errcheck
950 if err := eng.Setup(); err != nil {
951 runErr = err
952 return fmt.Errorf("setup: %w", err)
953 }
954 fmt.Fprintf(os.Stdout, "✅ workspace ready %s\n", eng.Wk.GetDir()) //nolint:errcheck
955
956 // Store the event in the artifact directory so the monitor can access it
957 artifactsDir := filepath.Join(eventDir, "artifacts")
958 if err := os.MkdirAll(artifactsDir, 0755); err != nil {
959 log.Error("create artifacts dir", "err", err)
960 } else {
961 if err := os.WriteFile(filepath.Join(artifactsDir, "event.json"), eventBytes, 0644); err != nil {
962 log.Error("write event", "err", err)
963 }
964 }
965
966 // Write attestation.json with runner/workspace provenance
967 hostname, _ := os.Hostname()
968 attestation := map[string]interface{}{
969 "runner": map[string]string{
970 "hostname": hostname,
971 "os": runtimeOS(),
972 "arch": runtimeArch(),
973 },
974 "provenance": map[string]string{
975 "repo": eventData.Name,
976 "branch": eventData.Branch,
977 "commit": eventData.Commit,
978 },
979 "workspace_checksum": eng.Wk.Checksum(),
980 }
981 attestationBytes, _ := json.Marshal(attestation)
982 if err := os.WriteFile(filepath.Join(artifactsDir, "attestation.json"), attestationBytes, 0644); err != nil {
983 log.Error("write attestation", "err", err)
984 }
985
986 manifest, err := eng.FindManifest()
987 if err != nil {
988 runErr = err
989 fmt.Fprintf(os.Stdout, "❌ %s\n\n", err) //nolint:errcheck
990 //nolint:errcheck
991 fmt.Fprint(os.Stdout, `Create a pico.sh script in your workspace root:
992
993 #!/usr/bin/env bash
994 set -euo pipefail
995
996 export ZMX_SESSION_PREFIX="${ZMX_SESSION_PREFIX:-ci.}"
997
998 # Run your CI steps as zmx sessions
999 zmx run lint -d <your lint command>
1000 zmx run test -d <your test command>
1001
1002 # Wait for all steps to complete
1003 zmx wait "*"
1004
1005 printf "\x1b[32msuccess!\x1b[0m\n"
1006
1007Each 'zmx run' spawns a parallel session. 'zmx wait "*"' blocks
1008until all child sessions finish.
1009
1010You can also run this script in isolation without the runner.
1011
1012See: https://github.com/picosh/pici
1013
1014`)
1015 return err
1016 }
1017 fmt.Fprintf(os.Stdout, "🔍 found %s\n", manifest) //nolint:errcheck
1018
1019 fmt.Fprint(os.Stdout, "🏃 launching sessions...\n") //nolint:errcheck
1020 if err := eng.Run(manifest); err != nil {
1021 runErr = err
1022 return fmt.Errorf("run: %w", err)
1023 }
1024
1025 fmt.Fprintln(os.Stdout, "✅ job launched") //nolint:errcheck
1026
1027 if cfg.Wait {
1028 if err := waitAndReportPlain(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
1029 runErr = err
1030 return fmt.Errorf("wait: %w", err)
1031 }
1032 return nil
1033 }
1034
1035 domain = getDomain(eventData.Type)
1036 session := fmt.Sprintf("%s.%s.%s.runner", domain, eventData.Name, jobID)
1037 fmt.Fprintf(os.Stdout, " zmx tail %s\n", session) //nolint:errcheck
1038 fmt.Fprintf(os.Stdout, " zmx history %s\n", session) //nolint:errcheck
1039 fmt.Fprintf(os.Stdout, " zmx attach %s\n", session) //nolint:errcheck
1040 return nil
1041}
1042
1043// waitAndReport polls the job's sessions until all complete, prints live
1044// interactive progress to stdout, updates stats.json, and outputs a final summary.
1045func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) error {
1046 cwd, _ := os.Getwd()
1047 branch := detectGitBranch(cwd)
1048 commit := detectGitCommit(cwd)
1049 return waitAndReportUI(cfg, log, name, jobID, eventType, branch, commit)
1050}
1051
1052func writeSummaryMarkdown(filePath string, repoName, jobID string, jobSessions []SessionInfo, known map[string]*sessionState, debugActive bool, debugTopic string) error {
1053 if filePath == "" {
1054 return nil
1055 }
1056
1057 exitCode, status := resolveJobExitCode(jobSessions)
1058 _, _, duration := computeJobTiming(jobSessions)
1059 statusIcon := map[string]string{"success": "✅", "failed": "❌"}[status]
1060
1061 var sb strings.Builder
1062 fmt.Fprintf(&sb, "### Pici CI Summary: `%s` (Job `%s`)\n\n", repoName, jobID)
1063 fmt.Fprintf(&sb, "**Status:** %s %s | **Exit Code:** `%d` | **Duration:** `%s`\n\n", statusIcon, strings.ToUpper(status), exitCode, duration)
1064
1065 if debugActive && debugTopic != "" {
1066 sb.WriteString("### 🔧 Remote Debug Active\n\n")
1067 sb.WriteString("To attach to the interactive shell session for this job:\n\n")
1068 fmt.Fprintf(&sb, "```bash\nssh -t pipe.pico.sh pipe %s\n```\n\n", debugTopic)
1069 }
1070
1071 sb.WriteString("| Session | Status | Exit Code | Duration |\n")
1072 sb.WriteString("| :--- | :--- | :--- | :--- |\n")
1073
1074 for _, s := range jobSessions {
1075 state := known[s.Short]
1076 sStatus := "unknown"
1077 sExit := "—"
1078 sDur := "—"
1079 sIcon := "❓"
1080 if state != nil {
1081 sStatus = state.status
1082 sExit = state.exitCode
1083 if sExit == "" {
1084 sExit = "—"
1085 }
1086 sDur = state.duration
1087 if sDur == "" {
1088 sDur = "—"
1089 }
1090 sIcon = map[string]string{"running": "🚀", "success": "✅", "failed": "❌"}[sStatus]
1091 }
1092 fmt.Fprintf(&sb, "| `%s` | %s %s | `%s` | %s |\n", s.Short, sIcon, sStatus, sExit, sDur)
1093 }
1094 sb.WriteString("\n")
1095
1096 // Include failed session details
1097 hasFailed := false
1098 for _, s := range jobSessions {
1099 state := known[s.Short]
1100 if state == nil || state.status != "failed" {
1101 continue
1102 }
1103 if !hasFailed {
1104 sb.WriteString("### Failed Session Details\n\n")
1105 hasFailed = true
1106 }
1107 history, err := fetchHistoryPlain(s.Name)
1108 if err != nil {
1109 history = fmt.Sprintf("(history unavailable: %v)", err)
1110 }
1111 fmt.Fprintf(&sb, "<details><summary><b>%s</b> (exit %s)</summary>\n\n```\n%s\n```\n\n</details>\n\n", s.Short, state.exitCode, strings.TrimSpace(history))
1112 }
1113
1114 f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
1115 if err != nil {
1116 return fmt.Errorf("open summary file: %w", err)
1117 }
1118 defer func() { _ = f.Close() }()
1119
1120 if _, err := f.WriteString(sb.String()); err != nil {
1121 return fmt.Errorf("write summary file: %w", err)
1122 }
1123 return nil
1124}
1125
1126// cleanSessionShort produces a readable short name from a full session name.
1127// ci.name.jobID.step.lint → lint
1128// ci.name.jobID.step.ci.name.jobID.runner → runner
1129func cleanSessionShort(name, prefix, repoName, jobID string) string {
1130 short := strings.TrimPrefix(name, prefix)
1131 // Strip "step." prefix added by child sessions
1132 short = strings.TrimPrefix(short, "step.")
1133 // Strip nested full prefix (e.g. runner named ci.name.jobID.runner or local.name.jobID.runner)
1134 nestedCI := "ci." + repoName + "." + jobID + "."
1135 short = strings.TrimPrefix(short, nestedCI)
1136 nestedLocal := "local." + repoName + "." + jobID + "."
1137 short = strings.TrimPrefix(short, nestedLocal)
1138 return short
1139}
1140
1141// runMonitor is a long-lived daemon that polls all ci.* zmx sessions,
1142// writes status as JSONL to stdout, stages artifacts, and syncs to destination.
1143// Logs go to stderr. Compose with shell tools to route status:
1144//
1145// pici monitor | ssh pipe.pico.sh "pub build.status -b=false"
1146// pici monitor | while read -r line; do curl -sd"$line" $WEBHOOK; done
1147// pici monitor > status.jsonl
1148//
1149// sessionState tracks the display state of a single session.
1150type sessionState struct {
1151 status string // "running", "success", "failed"
1152 exitCode string
1153 duration string
1154}
1155
1156// monitorJobState tracks display state for a single job across ticks.
1157type monitorJobState struct {
1158 sessionOrder []string // insertion order for deterministic output
1159 sessions map[string]*sessionState
1160 liveLines []string // last set of status lines printed (for overwrite)
1161}
1162
1163func runMonitor(cfg *Cfg) error {
1164 log := cfg.Logger.With("cmd", "monitor")
1165
1166 output := cfg.StatusOutput
1167 if output == nil {
1168 output = os.Stdout
1169 }
1170
1171 ticker := time.NewTicker(cfg.MonitorInterval)
1172 defer ticker.Stop()
1173
1174 // Optional GC ticker — runs garbage collection on a separate interval.
1175 var gcChan <-chan time.Time
1176 if cfg.GCInterval > 0 {
1177 gcTicker := time.NewTicker(cfg.GCInterval)
1178 defer gcTicker.Stop()
1179 gcChan = gcTicker.C
1180 }
1181
1182 // Track per-job display state across ticks (for human output)
1183 jobStates := make(map[string]*monitorJobState) // key: "name/jobID"
1184
1185 log.Debug("monitor started", "interval", cfg.MonitorInterval, "gc_interval", cfg.GCInterval, "artifact_dir", cfg.ArtifactDir)
1186 log.Debug("monitoring ci.* sessions for job status, writing status to stdout")
1187
1188 for {
1189 select {
1190 case <-cfg.Ctx.Done():
1191 log.Debug("context cancelled, stopping monitor")
1192 return cfg.Ctx.Err()
1193 case <-ticker.C:
1194 if err := monitorTick(cfg, log, output, jobStates); err != nil {
1195 log.Error("monitor tick", "err", err)
1196 }
1197 case <-gcChan:
1198 log.Debug("running periodic garbage collection")
1199 if err := runGC(cfg); err != nil {
1200 log.Error("periodic gc", "err", err)
1201 }
1202 }
1203 }
1204}
1205
1206// monitorTick performs a single monitoring pass over all ci.* sessions.
1207// renderJobRunning prints per-session status for a running job, using ANSI
1208// cursor control to overwrite previous lines.
1209func renderJobRunning(output io.Writer, name, jobID string, group []SessionInfo, duration string, jobStates map[string]*monitorJobState) {
1210 key := name + "/" + jobID
1211 state, ok := jobStates[key]
1212 if !ok {
1213 state = &monitorJobState{sessions: make(map[string]*sessionState)}
1214 jobStates[key] = state
1215 }
1216
1217 // Track new sessions
1218 for _, s := range group {
1219 if _, ok := state.sessions[s.Short]; !ok {
1220 state.sessions[s.Short] = &sessionState{}
1221 state.sessionOrder = append(state.sessionOrder, s.Short)
1222 }
1223 }
1224
1225 // Update session states
1226 for _, s := range group {
1227 ss := state.sessions[s.Short]
1228 if s.Ended == "" {
1229 ss.status = "running"
1230 created, _ := strconv.ParseInt(s.Created, 10, 64)
1231 ss.duration = fmtDurationTs(created, time.Now().Unix())
1232 } else {
1233 if s.ExitCode == "0" {
1234 ss.status = "success"
1235 } else {
1236 ss.status = "failed"
1237 ss.exitCode = s.ExitCode
1238 }
1239 ss.duration = fmtDuration(s.Created, s.Ended)
1240 }
1241 }
1242
1243 // Build status lines
1244 lines := make([]string, 0, len(state.sessionOrder))
1245 for _, short := range state.sessionOrder {
1246 ss := state.sessions[short]
1247 icon := map[string]string{"running": "🚀", "success": "✅", "failed": "❌"}[ss.status]
1248 detail := ""
1249 if ss.status == "failed" && ss.exitCode != "" {
1250 detail = fmt.Sprintf(", exit %s", ss.exitCode)
1251 }
1252 if ss.duration != "" && ss.duration != "—" {
1253 detail += fmt.Sprintf(" (%s)", ss.duration)
1254 }
1255 lines = append(lines, fmt.Sprintf(" %-12s %s %s%s", short, icon, ss.status, detail))
1256 }
1257
1258 // Overwrite previous lines or print fresh
1259 if len(state.liveLines) > 0 {
1260 for range len(state.liveLines) {
1261 fmt.Fprint(output, "\033[A") //nolint:errcheck
1262 }
1263 for i, line := range lines {
1264 if i > 0 {
1265 fmt.Fprint(output, "\n") //nolint:errcheck
1266 }
1267 fmt.Fprint(output, line+"\033[K") //nolint:errcheck
1268 }
1269 fmt.Fprint(output, "\n") //nolint:errcheck
1270 } else {
1271 fmt.Fprintf(output, " %-12s %s %s\n", name, "🚀", "running") //nolint:errcheck
1272 for _, line := range lines {
1273 fmt.Fprintln(output, line) //nolint:errcheck
1274 }
1275 }
1276 state.liveLines = lines
1277}
1278
1279// renderJobFinal prints the final status for a completed job.
1280func renderJobFinal(output io.Writer, name, jobID string, group []SessionInfo, duration, status string, success bool, workspace, artifactDir string) {
1281 icon := map[string]string{"success": "✅", "failed": "❌"}[status]
1282 fmt.Fprintf(output, " %-12s %s %s (%s)\n", name, icon, status, duration) //nolint:errcheck
1283
1284 // Per-session summary
1285 for _, s := range group {
1286 icon := "✅"
1287 if s.ExitCode != "0" {
1288 icon = "❌"
1289 }
1290 dur := fmtDuration(s.Created, s.Ended)
1291 fmt.Fprintf(output, " %-12s %s done (%s)\n", s.Short, icon, dur) //nolint:errcheck
1292 }
1293
1294 // Context info
1295 fmt.Fprint(output, "\n") //nolint:errcheck
1296 if workspace != "" {
1297 fmt.Fprintf(output, " workspace: %s\n", workspace) //nolint:errcheck
1298 }
1299 artifactPath := filepath.Join(artifactDir, name, jobID)
1300 fmt.Fprintf(output, " artifacts: %s\n", artifactPath) //nolint:errcheck
1301 fmt.Fprint(output, "\n") //nolint:errcheck
1302}
1303
1304func monitorTick(cfg *Cfg, log *slog.Logger, output io.Writer, jobStates map[string]*monitorJobState) error {
1305 // a. zmx list → filter ci.* sessions
1306 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
1307 if err != nil {
1308 return fmt.Errorf("zmx list: %w", err)
1309 }
1310 sessions := parseZMXList(string(listOutput))
1311 ciSessions := filterCISessions(cfg, sessions)
1312
1313 if len(ciSessions) == 0 {
1314 log.Debug("no ci.* sessions found")
1315 return nil
1316 }
1317
1318 log.Debug("found ci sessions", "count", len(ciSessions))
1319
1320 // b. Group by job prefix: ci.<name>.<jobID>.
1321 groups := groupSessionsByJob(ciSessions)
1322
1323 // c. Process each job group
1324 for prefix, group := range groups {
1325 name, jobID := parseJobPrefix(prefix)
1326 if name == "" {
1327 continue
1328 }
1329
1330 log := log.With("repo", name, "job_id", jobID)
1331
1332 // Fetch and stage history for every session at each tick,
1333 // not just when the job completes. This gives live progress
1334 // snapshots while the job is running.
1335 for _, s := range group {
1336 // Determine session status and timing
1337 sessionStatus := "running"
1338 sessionDuration := fmtDuration(s.Created, fmt.Sprintf("%d", time.Now().Unix()))
1339 sessionExitCode := ""
1340 if s.Ended != "" {
1341 sessionDuration = fmtDuration(s.Created, s.Ended)
1342 if s.ExitCode == "0" {
1343 sessionStatus = "success"
1344 sessionExitCode = "0"
1345 } else {
1346 sessionStatus = "failed"
1347 sessionExitCode = s.ExitCode
1348 }
1349 }
1350
1351 html, err := fetchHistoryHTML(s.Name, name, jobID, sessionStatus, sessionDuration, sessionExitCode)
1352 if err != nil {
1353 log.Error("fetch history html", "session", s.Name, "err", err)
1354 continue
1355 }
1356 if err := stageArtifact(cfg.ArtifactDir, name, jobID, s.Short, html, ".html"); err != nil {
1357 log.Error("stage html artifact", "session", s.Short, "err", err)
1358 }
1359
1360 plain, err := fetchHistoryPlain(s.Name)
1361 if err != nil {
1362 log.Error("fetch history plain", "session", s.Name, "err", err)
1363 continue
1364 }
1365 if err := stageArtifact(cfg.ArtifactDir, name, jobID, s.Short, plain, ".txt"); err != nil {
1366 log.Error("stage txt artifact", "session", s.Short, "err", err)
1367 }
1368 }
1369
1370 // Generate and stage job index landing pages
1371 indexHTML, indexTXT := generateJobIndex(cfg.ArtifactDir, name, jobID, group)
1372 if err := stageArtifact(cfg.ArtifactDir, name, jobID, "index", indexHTML, ".html"); err != nil {
1373 log.Error("stage index.html", "err", err)
1374 }
1375 if err := stageArtifact(cfg.ArtifactDir, name, jobID, "index", indexTXT, ".txt"); err != nil {
1376 log.Error("stage index.txt", "err", err)
1377 }
1378 // Stage shared CSS
1379 if styles, err := loadStyles(); err == nil {
1380 if err := stageArtifact(cfg.ArtifactDir, name, jobID, "styles", styles, ".css"); err != nil {
1381 log.Error("stage styles.css", "err", err)
1382 }
1383 }
1384
1385 // Load event to get artifact destination
1386 eventData, _ := loadEvent(cfg.ArtifactDir, name, jobID)
1387
1388 // Compute job-level timing from session timestamps
1389 startedAt, endedAt, duration := computeJobTiming(group)
1390
1391 if isJobComplete(group) {
1392 // Check sentinel — publish terminal status exactly once
1393 sentinel := filepath.Join(cfg.ArtifactDir, name, jobID, "artifacts", "published.json")
1394 log.Debug("checking completion", "all_completed", true, "sentinel", sentinel)
1395 if _, err := os.Stat(sentinel); err == nil {
1396 log.Debug("terminal status already published, skipping", "sentinel", sentinel)
1397 continue
1398 }
1399
1400 log.Debug("job completed, publishing final status", "sessions", len(group))
1401 exitCode, status := resolveJobExitCode(group)
1402 log.Info("job finished", "status", status, "exit_code", exitCode)
1403
1404 if cfg.HumanOutput {
1405 renderJobFinal(output, name, jobID, group, duration, status, exitCode == 0, eventData.Workspace, cfg.ArtifactDir)
1406 } else {
1407 payload := StatusPayload{
1408 Timestamp: time.Now().UTC().Format(time.RFC3339),
1409 Name: name,
1410 JobID: jobID,
1411 Status: status,
1412 ExitCode: &exitCode,
1413 Duration: duration,
1414 StartedAt: startedAt,
1415 EndedAt: endedAt,
1416 SessionCount: len(group),
1417 Sessions: group,
1418 }
1419 if err := publishStatus(output, payload); err != nil {
1420 log.Error("publish final status", "err", err)
1421 }
1422 }
1423 log.Info("writing sentinel", "repo", name, "job_id", jobID)
1424 if err := writePublishedSentinel(cfg.ArtifactDir, name, jobID, status, exitCode); err != nil {
1425 log.Error("write published sentinel", "err", err)
1426 }
1427 // Regenerate index.html now that published.json exists, so the artifact list includes it.
1428 indexHTML, indexTXT := generateJobIndex(cfg.ArtifactDir, name, jobID, group)
1429 if err := stageArtifact(cfg.ArtifactDir, name, jobID, "index", indexHTML, ".html"); err != nil {
1430 log.Error("stage index.html", "err", err)
1431 }
1432 if err := stageArtifact(cfg.ArtifactDir, name, jobID, "index", indexTXT, ".txt"); err != nil {
1433 log.Error("stage index.txt", "err", err)
1434 }
1435 if err := syncJobArtifacts(cfg, name, jobID, log); err != nil {
1436 log.Error("sync artifacts", "err", err)
1437 }
1438 } else {
1439 log.Debug("job still running", "sessions", len(group))
1440 // Sync in-progress jobs every tick using the sessions we already know about.
1441 if err := syncJobArtifacts(cfg, name, jobID, log); err != nil {
1442 log.Error("sync artifacts", "err", err)
1443 }
1444 // Publish running status only when --include-running is set
1445 if cfg.IncludeRunning {
1446 if cfg.HumanOutput {
1447 renderJobRunning(output, name, jobID, group, duration, jobStates)
1448 } else {
1449 payload := StatusPayload{
1450 Timestamp: time.Now().UTC().Format(time.RFC3339),
1451 Name: name,
1452 JobID: jobID,
1453 Status: "running",
1454 ExitCode: nil,
1455 Duration: duration,
1456 StartedAt: startedAt,
1457 SessionCount: len(group),
1458 Sessions: group,
1459 }
1460 if err := publishStatus(output, payload); err != nil {
1461 log.Error("publish status", "err", err)
1462 }
1463 }
1464 }
1465 }
1466 }
1467
1468 return nil
1469}
1470
1471// loadEvent reads the event.json for a job from the artifact directory.
1472func loadEvent(dir, name, jobID string) (Event, error) {
1473 var event Event
1474 data, err := os.ReadFile(filepath.Join(dir, name, jobID, "artifacts", "event.json"))
1475 if err != nil {
1476 return event, err
1477 }
1478 return event, json.Unmarshal(data, &event)
1479}
1480
1481// computeJobTiming derives started_at, ended_at, and duration from session timestamps.
1482// started_at is the earliest session creation time, ended_at is the latest session end time.
1483// For running jobs, ended_at is empty and duration is computed against now.
1484func computeJobTiming(sessions []SessionInfo) (startedAt, endedAt, duration string) {
1485 var earliestCreated, latestEnded int64
1486 hasCreated, hasEnded := false, false
1487
1488 for _, s := range sessions {
1489 if s.Created != "" {
1490 var c int64
1491 if _, err := fmt.Sscanf(s.Created, "%d", &c); err == nil {
1492 if !hasCreated || c < earliestCreated {
1493 earliestCreated = c
1494 }
1495 hasCreated = true
1496 }
1497 }
1498 if s.Ended != "" {
1499 var e int64
1500 if _, err := fmt.Sscanf(s.Ended, "%d", &e); err == nil {
1501 if !hasEnded || e > latestEnded {
1502 latestEnded = e
1503 }
1504 hasEnded = true
1505 }
1506 }
1507 }
1508
1509 if hasCreated {
1510 startedAt = time.Unix(earliestCreated, 0).UTC().Format(time.RFC3339)
1511 }
1512 if hasEnded {
1513 endedAt = time.Unix(latestEnded, 0).UTC().Format(time.RFC3339)
1514 }
1515
1516 if hasCreated && hasEnded {
1517 duration = fmtDurationTs(earliestCreated, latestEnded)
1518 } else if hasCreated {
1519 duration = fmtDurationTs(earliestCreated, time.Now().Unix())
1520 }
1521
1522 return startedAt, endedAt, duration
1523}
1524
1525// fmtDurationTs formats the duration between two unix timestamps.
1526func fmtDurationTs(started, ended int64) string {
1527 d := time.Duration(ended-started) * time.Second
1528 if d < 0 {
1529 return "—"
1530 }
1531 if d >= time.Minute {
1532 return fmt.Sprintf("%dm%ds", d/time.Minute, d%time.Minute/time.Second)
1533 }
1534 secs := float64(d) / float64(time.Second)
1535 if secs >= 10 {
1536 return fmt.Sprintf("%ds", int(secs))
1537 }
1538 return fmt.Sprintf("%.1fs", secs)
1539}
1540
1541// filterCISessions returns only sessions matching cfg.SessionPrefix (default "ci.").
1542func filterCISessions(cfg *Cfg, sessions []SessionInfo) []SessionInfo {
1543 prefix := "ci."
1544 if cfg != nil && cfg.SessionPrefix != "" {
1545 prefix = cfg.SessionPrefix
1546 }
1547 if !strings.HasSuffix(prefix, ".") {
1548 prefix = prefix + "."
1549 }
1550 var filtered []SessionInfo
1551 for _, s := range sessions {
1552 if strings.HasPrefix(s.Name, prefix) {
1553 filtered = append(filtered, s)
1554 }
1555 }
1556 return filtered
1557}
1558
1559// groupSessionsByJob groups sessions by their job prefix (ci.<name>.<jobID>.).
1560func groupSessionsByJob(sessions []SessionInfo) map[string][]SessionInfo {
1561 groups := make(map[string][]SessionInfo)
1562 for _, s := range sessions {
1563 prefix := extractJobPrefix(s.Name)
1564 if prefix == "" {
1565 continue
1566 }
1567 // Set the Short name
1568 name, jobID := parseJobPrefix(prefix)
1569 s.Short = cleanSessionShort(s.Name, prefix, name, jobID)
1570 groups[prefix] = append(groups[prefix], s)
1571 }
1572 return groups
1573}
1574
1575// parseJobPrefix extracts name and jobID from a prefix like "ci.<name>.<jobID>.".
1576func parseJobPrefix(prefix string) (name, jobID string) {
1577 // ci.name.jobID. -> ["ci", "name", "jobID", ""]
1578 parts := strings.Split(prefix, ".")
1579 if len(parts) < 4 {
1580 return "", ""
1581 }
1582 return parts[1], parts[2]
1583}
1584
1585// resolveJobExitCode determines the job's exit code from its sessions.
1586// Defensive: if any child session failed, the job failed regardless of the
1587// runner's exit code. This protects against bad pico.sh scripts that exit 0
1588// without waiting for children.
1589func resolveJobExitCode(sessions []SessionInfo) (int, string) {
1590 var runnerExit *int
1591 var worstChild *int // highest non-zero child exit code
1592
1593 for _, s := range sessions {
1594 if s.ExitCode == "" {
1595 continue
1596 }
1597 var code int
1598 if _, err := fmt.Sscanf(s.ExitCode, "%d", &code); err != nil {
1599 continue
1600 }
1601
1602 if strings.HasSuffix(s.Name, ".runner") {
1603 runnerExit = &code
1604 } else if code != 0 {
1605 if worstChild == nil || code > *worstChild {
1606 worstChild = &code
1607 }
1608 }
1609 }
1610
1611 // Any child failure overrides the runner — defensive against bad scripts
1612 if worstChild != nil {
1613 return *worstChild, "failed"
1614 }
1615 if runnerExit != nil && *runnerExit != 0 {
1616 return *runnerExit, "failed"
1617 }
1618 return 0, "success"
1619}
1620
1621func generateJobID(name, workspace string) string {
1622 return jobIDFor(name, workspace, time.Now().UnixNano())
1623}
1624
1625// resolveJobID returns the producer-provided job ID if set and valid,
1626// otherwise generates one from name + workspace + timestamp.
1627func resolveJobID(name, workspace, providedID string) string {
1628 if providedID != "" && validateJobID(providedID) {
1629 return providedID
1630 }
1631 return generateJobID(name, workspace)
1632}
1633
1634// validateJobID checks that the ID is safe to use in session names and paths.
1635// Constraints: alphanumeric + hyphens only, no dots (breaks parseJobPrefix),
1636// 1-16 chars to keep session names reasonable.
1637func validateJobID(id string) bool {
1638 if len(id) == 0 || len(id) > 16 {
1639 return false
1640 }
1641 for _, r := range id {
1642 if (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') && (r < '0' || r > '9') && r != '-' {
1643 return false
1644 }
1645 }
1646 return true
1647}
1648
1649func runtimeOS() string {
1650 return runtime.GOOS
1651}
1652
1653func runtimeArch() string {
1654 return runtime.GOARCH
1655}
1656
1657func jobIDFor(name, workspace string, ts int64) string {
1658 h := sha256.Sum256([]byte(name + workspace + fmt.Sprintf("%d", ts)))
1659 return fmt.Sprintf("%x", h[:4])
1660}
1661
1662func parseZMXList(output string) []SessionInfo {
1663 var sessions []SessionInfo
1664 lines := strings.Split(strings.TrimSpace(output), "\n")
1665 for _, line := range lines {
1666 line = strings.TrimSpace(line)
1667 if line == "" {
1668 continue
1669 }
1670 // Strip leading arrow/space prefix
1671 line = strings.TrimPrefix(line, "→ ")
1672 line = strings.TrimSpace(line)
1673
1674 fields := strings.FieldsFunc(line, func(r rune) bool {
1675 return r == '\t'
1676 })
1677
1678 var si SessionInfo
1679 for _, field := range fields {
1680 parts := strings.SplitN(field, "=", 2)
1681 if len(parts) != 2 {
1682 continue
1683 }
1684 switch parts[0] {
1685 case "name":
1686 si.Name = parts[1]
1687 case "pid":
1688 si.PID = parts[1]
1689 case "clients":
1690 si.Clients = parts[1]
1691 case "created":
1692 si.Created = parts[1]
1693 case "start_dir":
1694 si.StartDir = parts[1]
1695 case "ended":
1696 si.Ended = parts[1]
1697 case "exit_code":
1698 si.ExitCode = parts[1]
1699 }
1700 }
1701 if si.Name != "" {
1702 sessions = append(sessions, si)
1703 }
1704 }
1705 return sessions
1706}
1707
1708// loadStyles reads the shared CSS from the embedded template filesystem.
1709func loadStyles() (string, error) {
1710 data, err := fs.ReadFile(tmplFS, "tmpl/styles.css")
1711 if err != nil {
1712 return "", fmt.Errorf("read styles: %w", err)
1713 }
1714 return string(data), nil
1715}
1716
1717// SessionArtifactData holds the data for rendering a session HTML artifact.
1718type SessionArtifactData struct {
1719 SessionName string
1720 SessionShort string
1721 SessionStatus string
1722 JobName string
1723 JobID string
1724 Duration string
1725 ExitCode string
1726 Content template.HTML
1727}
1728
1729// fetchHistoryHTML fetches session history from zmx and wraps it in a full HTML document.
1730func fetchHistoryHTML(sessionName, jobName, jobID, status, duration, exitCode string) (string, error) {
1731 cmd := exec.Command("zmx", "history", sessionName, "--html")
1732 output, err := cmd.Output()
1733 if err != nil {
1734 return "", err
1735 }
1736
1737 prefix := "ci." + jobName + "." + jobID + "."
1738 shortName := strings.TrimPrefix(sessionName, prefix)
1739 shortName = strings.TrimPrefix(shortName, "step.")
1740
1741 data := SessionArtifactData{
1742 SessionName: sessionName,
1743 SessionShort: shortName,
1744 SessionStatus: status,
1745 JobName: jobName,
1746 JobID: jobID,
1747 Duration: duration,
1748 ExitCode: exitCode,
1749 Content: template.HTML(string(output)),
1750 }
1751
1752 tmpl, err := template.ParseFS(tmplFS, "tmpl/session.html")
1753 if err != nil {
1754 return "", fmt.Errorf("parse template: %w", err)
1755 }
1756
1757 var buf strings.Builder
1758 if err := tmpl.ExecuteTemplate(&buf, "session.html", data); err != nil {
1759 return "", fmt.Errorf("execute template: %w", err)
1760 }
1761
1762 return buf.String(), nil
1763}
1764
1765func fetchHistoryPlain(sessionName string) (string, error) {
1766 cmd := exec.Command("zmx", "history", sessionName, "--plain")
1767 output, err := cmd.Output()
1768 if err != nil {
1769 return "", err
1770 }
1771 return string(output), nil
1772}
1773
1774func publishStatus(w io.Writer, payload StatusPayload) error {
1775 data, err := json.Marshal(payload)
1776 if err != nil {
1777 return err
1778 }
1779 if _, err := w.Write(append(data, '\n')); err != nil {
1780 return err
1781 }
1782 if f, ok := w.(interface{ Flush() error }); ok {
1783 _ = f.Flush()
1784 } else if f, ok := w.(*os.File); ok {
1785 _ = f.Sync()
1786 }
1787 return nil
1788}
1789
1790func stageArtifact(dir, name, jobID, short, content, ext string) error {
1791 path := filepath.Join(dir, name, jobID, short+ext)
1792 if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
1793 return err
1794 }
1795 return os.WriteFile(path, []byte(content), 0644)
1796}
1797
1798// syncJobArtifacts syncs artifacts for a single job.
1799func syncJobArtifacts(cfg *Cfg, repoName, jobID string, log *slog.Logger) error {
1800 jobDir := filepath.Join(cfg.ArtifactDir, repoName, jobID)
1801 eventPath := filepath.Join(jobDir, "artifacts", "event.json")
1802 data, err := os.ReadFile(eventPath)
1803 if err != nil {
1804 if os.IsNotExist(err) {
1805 log.Debug("event.json not found, skipping sync", "repo", repoName, "job_id", jobID)
1806 return nil
1807 }
1808 return fmt.Errorf("read event: %w", err)
1809 }
1810 var event Event
1811 if err := json.Unmarshal(data, &event); err != nil || event.ArtifactDest == "" {
1812 return fmt.Errorf("invalid event")
1813 }
1814 log.Debug("syncing artifacts", "repo", repoName, "job_id", jobID, "dest", event.ArtifactDest)
1815 var sshArgs string
1816 if cfg.KeyLocation != "" {
1817 certFile := ""
1818 if cfg.CertificateLocation != "" {
1819 certFile = fmt.Sprintf(" -o CertificateFile %s", cfg.CertificateLocation)
1820 }
1821 sshArgs = fmt.Sprintf("-F ~/.ssh/config -i %s%s", cfg.KeyLocation, certFile)
1822 }
1823 // Source is jobDir (with trailing slash) so rsync copies
1824 // the job's artifact contents directly into event.ArtifactDest.
1825 dest := event.ArtifactDest
1826 if !strings.HasSuffix(dest, "/") {
1827 dest += "/"
1828 }
1829 srcDir := strings.TrimSuffix(jobDir, "/") + "/"
1830 var cmd *exec.Cmd
1831 if sshArgs != "" {
1832 cmd = exec.Command("rsync", "-e", sshArgs, "-rv", srcDir, dest)
1833 } else {
1834 cmd = exec.Command("rsync", "-rv", srcDir, dest)
1835 }
1836 rsyncCmd := fmt.Sprintf("rsync %s %s %s",
1837 strings.TrimLeft(cmd.Args[1], "-"),
1838 srcDir, dest)
1839 log.Info("rsync", "cmd", rsyncCmd)
1840 return runCmd(cmd, log)
1841}
1842
1843func allCompleted(sessions []SessionInfo) bool {
1844 for _, s := range sessions {
1845 if s.Ended == "" {
1846 return false
1847 }
1848 }
1849 return true
1850}
1851
1852// isJobComplete returns true if all sessions for a job have completed
1853// AND a runner session exists and has completed. This ensures monitor and wait
1854// do not mark a job complete while pico.sh (the runner) is still executing and
1855// potentially spawning new tasks.
1856func isJobComplete(sessions []SessionInfo) bool {
1857 if len(sessions) == 0 {
1858 return false
1859 }
1860 hasRunner := false
1861 for _, s := range sessions {
1862 if s.Ended == "" {
1863 return false
1864 }
1865 if strings.HasSuffix(s.Name, ".runner") {
1866 hasRunner = true
1867 }
1868 }
1869 return hasRunner
1870}
1871
1872func runCmd(cmd *exec.Cmd, log *slog.Logger) error {
1873 stdout, err := cmd.StdoutPipe()
1874 if err != nil {
1875 return err
1876 }
1877
1878 stderr, err := cmd.StderrPipe()
1879 if err != nil {
1880 return err
1881 }
1882
1883 if err := cmd.Start(); err != nil {
1884 return err
1885 }
1886
1887 go func() {
1888 scanner := bufio.NewScanner(stdout)
1889 for scanner.Scan() {
1890 log.Debug("cmd stdout", "text", scanner.Text())
1891 }
1892 }()
1893
1894 go func() {
1895 scanner := bufio.NewScanner(stderr)
1896 for scanner.Scan() {
1897 log.Error("cmd stderr", "text", scanner.Text())
1898 }
1899 }()
1900
1901 return cmd.Wait()
1902}
1903
1904// runCancel reads an event from stdin and cancels any running job with matching name+type.
1905func runCancel(cfg *Cfg) error {
1906 log := cfg.Logger.With("cmd", "cancel")
1907
1908 // Read event from stdin
1909 scanner := bufio.NewScanner(os.Stdin)
1910 if !scanner.Scan() {
1911 return fmt.Errorf("no input on stdin")
1912 }
1913
1914 var event Event
1915 if err := json.Unmarshal([]byte(scanner.Text()), &event); err != nil {
1916 return fmt.Errorf("unmarshal event: %w", err)
1917 }
1918
1919 log = log.With("repo", event.Name, "type", event.Type)
1920 log.Info("cancelling running jobs for repo")
1921
1922 cancelRunningJobs(cfg, log, event.Name)
1923 return nil
1924}
1925
1926// cancelRunningJobs finds and cancels all running jobs for a given repo name.
1927// It kills the runner sessions (which cascades to children). The monitor will
1928// detect the ended sessions and publish cancelled status on its next tick.
1929func cancelRunningJobs(cfg *Cfg, log *slog.Logger, name string) {
1930 runnerSessions, _ := findRunningJobs(name)
1931 if len(runnerSessions) == 0 {
1932 log.Debug("no running jobs to cancel")
1933 return
1934 }
1935
1936 log.Debug("found running jobs to cancel", "count", len(runnerSessions))
1937
1938 for _, runnerName := range runnerSessions {
1939 jobID := extractJobID(runnerName)
1940 log := log.With("job_id", jobID)
1941
1942 log.Debug("cancelling job", "runner", runnerName)
1943 if err := killSessions([]string{runnerName}); err != nil {
1944 log.Error("kill runner session", "err", err)
1945 continue
1946 }
1947 log.Debug("cancelled runner session")
1948 }
1949}
1950
1951// findRunningJobs finds all active runner sessions for a given name.
1952// Returns runner session names and all sessions for reference.
1953func findRunningJobs(name string) ([]string, []SessionInfo) {
1954 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
1955 if err != nil {
1956 return nil, nil
1957 }
1958
1959 sessions := parseZMXList(string(listOutput))
1960 var runners []string
1961 for _, s := range sessions {
1962 // Match ci.<name>.*.runner sessions that are still active (no ended)
1963 if (strings.HasPrefix(s.Name, "ci."+name+".") || strings.HasPrefix(s.Name, "local."+name+".")) && strings.HasSuffix(s.Name, ".runner") && s.Ended == "" {
1964 runners = append(runners, s.Name)
1965 }
1966 }
1967 return runners, sessions
1968}
1969
1970// extractJobID extracts the jobID from a runner session name like ci.<name>.<jobID>.runner.
1971func extractJobID(runnerName string) string {
1972 // ci.<name>.<jobID>.runner
1973 // Remove "ci." prefix and ".runner" suffix, then take the part after the first "."
1974 name := strings.TrimSuffix(runnerName, ".runner")
1975 name = strings.TrimPrefix(name, "ci.")
1976 name = strings.TrimPrefix(name, "local.")
1977 // name is now <name>.<jobID>, take the jobID part
1978 parts := strings.SplitN(name, ".", 2)
1979 if len(parts) == 2 {
1980 return parts[1]
1981 }
1982 return ""
1983}
1984
1985// killSessions kills zmx sessions by name.
1986func killSessions(names []string) error {
1987 if len(names) == 0 {
1988 return nil
1989 }
1990 args := append([]string{"kill"}, names...)
1991 cmd := exec.Command("zmx", args...)
1992 output, err := cmd.CombinedOutput()
1993 if err != nil {
1994 return fmt.Errorf("zmx kill: %s: %w", string(output), err)
1995 }
1996 return nil
1997}
1998
1999// cancelJobSessions finds and kills all active zmx sessions with the given prefix.
2000func cancelJobSessions(prefix string) {
2001 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
2002 if err != nil {
2003 return
2004 }
2005 sessions := parseZMXList(string(listOutput))
2006 var toKill []string
2007 for _, s := range sessions {
2008 if strings.HasPrefix(s.Name, prefix) && s.Ended == "" {
2009 toKill = append(toKill, s.Name)
2010 }
2011 }
2012 if len(toKill) > 0 {
2013 _ = killSessions(toKill)
2014 }
2015}
2016
2017// findSessionsForGC identifies finished or running sessions older than the retention cutoff (3 hours) matching given prefixes.
2018// It checks whether a session's job has an active session; active jobs are preserved unless their sessions have expired.
2019// Finished sessions are kept until they are older than the cutoff so users can inspect/debug test failures.
2020func findSessionsForGC(sessions []SessionInfo, prefixes []string, now time.Time) []string {
2021 cutoff := now.Add(-3 * time.Hour).Unix()
2022
2023 // Track which job prefixes have ANY active (running) session.
2024 // As long as any session for a job is still running (Ended == ""),
2025 // the job is active and none of its sessions should be garbage collected unless expired.
2026 activeJobPrefixes := make(map[string]bool)
2027 for _, s := range sessions {
2028 if s.Ended == "" {
2029 prefix := extractJobPrefix(s.Name)
2030 if prefix != "" {
2031 activeJobPrefixes[prefix] = true
2032 }
2033 }
2034 }
2035
2036 var toKill []string
2037 for _, s := range sessions {
2038 matched := false
2039 for _, prefix := range prefixes {
2040 if strings.HasPrefix(s.Name, prefix) {
2041 matched = true
2042 break
2043 }
2044 }
2045 if !matched {
2046 continue
2047 }
2048
2049 // If the session belongs to a job with an active session, do not kill it
2050 // unless it has expired (> 3 hours old).
2051 jobPrefix := extractJobPrefix(s.Name)
2052 if jobPrefix != "" && activeJobPrefixes[jobPrefix] {
2053 if s.Created != "" {
2054 var created int64
2055 if _, err := fmt.Sscanf(s.Created, "%d", &created); err != nil {
2056 continue
2057 }
2058 if created < cutoff {
2059 toKill = append(toKill, s.Name)
2060 }
2061 }
2062 continue
2063 }
2064
2065 // For completed/inactive jobs, only kill sessions if they ended or were created before the cutoff
2066 if s.Ended != "" {
2067 var ended int64
2068 if _, err := fmt.Sscanf(s.Ended, "%d", &ended); err == nil {
2069 if ended < cutoff {
2070 toKill = append(toKill, s.Name)
2071 }
2072 continue
2073 }
2074 }
2075
2076 if s.Created != "" {
2077 var created int64
2078 if _, err := fmt.Sscanf(s.Created, "%d", &created); err != nil {
2079 continue
2080 }
2081 if created < cutoff {
2082 toKill = append(toKill, s.Name)
2083 }
2084 }
2085 }
2086 return toKill
2087}
2088
2089// runGC kills finished zmx sessions or sessions older than 3 hours.
2090func runGC(cfg *Cfg) error {
2091 log := cfg.Logger.With("cmd", "gc")
2092 log.Debug("running garbage collection")
2093
2094 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
2095 if err != nil {
2096 return fmt.Errorf("zmx list: %w", err)
2097 }
2098
2099 sessions := parseZMXList(string(listOutput))
2100
2101 prefixes := []string{"ci.", "local."}
2102 if cfg != nil && cfg.SessionPrefix != "" {
2103 p := cfg.SessionPrefix
2104 if !strings.HasSuffix(p, ".") {
2105 p += "."
2106 }
2107 prefixes = append(prefixes, p)
2108 }
2109
2110 toKill := findSessionsForGC(sessions, prefixes, time.Now())
2111
2112 if len(toKill) == 0 {
2113 log.Debug("no sessions to garbage collect")
2114 return nil
2115 }
2116
2117 if err := killSessions(toKill); err != nil {
2118 return fmt.Errorf("kill sessions: %w", err)
2119 }
2120
2121 log.Debug("garbage collection complete", "killed", len(toKill))
2122 return nil
2123}
2124
2125// extractJobPrefix extracts the job prefix from a session name.
2126// ci.<name>.<jobID>.<step> -> ci.<name>.<jobID>.
2127func extractJobPrefix(sessionName string) string {
2128 // Extract the job key ci.<name>.<jobID> from session names:
2129 // ci.name.jobID.runner (4 parts) → ci.name.jobID.
2130 // ci.name.jobID.step.lint (5 parts) → ci.name.jobID.
2131 parts := strings.Split(sessionName, ".")
2132 if len(parts) < 4 {
2133 return ""
2134 }
2135 return parts[0] + "." + parts[1] + "." + parts[2] + "."
2136}
2137
2138// sessionRow holds display info for a single session in the job index.
2139type sessionRow struct {
2140 Name string
2141 Short string
2142 Status string
2143 ExitCode string
2144 Started string
2145 Ended string
2146 Duration string
2147}
2148
2149// artifactRow holds display info for non-session artifacts (e.g., event.json, workspace.tar).
2150type artifactRow struct {
2151 Name string
2152 Size string
2153 ModTime string
2154}
2155
2156// formatFileSize returns a human-readable file size string.
2157func formatFileSize(bytes int64) string {
2158 const unit = 1024
2159 if bytes < unit {
2160 return fmt.Sprintf("%d B", bytes)
2161 }
2162 div, exp := int64(unit), 0
2163 for n := bytes / unit; n >= unit; n /= unit {
2164 div *= unit
2165 exp++
2166 }
2167 units := []string{"KB", "MB", "GB", "TB"}
2168 return fmt.Sprintf("%.1f %s", float64(bytes)/float64(div), units[exp])
2169}
2170
2171// formatTimestamp converts a unix timestamp to a human-readable format.
2172func formatTimestamp(ts string) string {
2173 if ts == "" {
2174 return "—"
2175 }
2176 var t int64
2177 if _, err := fmt.Sscanf(ts, "%d", &t); err != nil {
2178 return "—"
2179 }
2180 return time.Unix(t, 0).UTC().Format("2006-01-02 15:04:05")
2181}
2182
2183// generateJobIndex produces HTML and plain-text index pages listing all
2184// sessions for a job with links and metadata (status, exit code, ended at).
2185// It also includes any other artifacts in the job directory (e.g., event.json, attestation.json, workspace.tar).
2186func generateJobIndex(artifactDir, name, jobID string, sessions []SessionInfo) (htmlContent, txtContent string) {
2187 // Sort sessions by Created (STARTED) timestamp
2188 sort.Slice(sessions, func(i, j int) bool {
2189 ci, _ := strconv.ParseInt(sessions[i].Created, 10, 64)
2190 cj, _ := strconv.ParseInt(sessions[j].Created, 10, 64)
2191 return ci < cj
2192 })
2193
2194 rows := make([]sessionRow, 0, len(sessions))
2195 for _, s := range sessions {
2196 row := sessionRow{
2197 Name: s.Name,
2198 Short: s.Short,
2199 Started: s.Created,
2200 }
2201 if s.Ended == "" {
2202 row.Status = "running"
2203 row.Duration = fmtDuration(s.Created, fmt.Sprintf("%d", time.Now().Unix()))
2204 } else if s.ExitCode == "0" {
2205 row.Status = "success"
2206 row.ExitCode = "0"
2207 } else {
2208 row.Status = "failed"
2209 row.ExitCode = s.ExitCode
2210 }
2211 row.Ended = s.Ended
2212 row.Duration = fmtDuration(s.Created, s.Ended)
2213 rows = append(rows, row)
2214 }
2215
2216 // Gather artifacts by scanning the artifacts/ subfolder
2217 var artifacts []artifactRow
2218 artifactsDir := filepath.Join(artifactDir, name, jobID, "artifacts")
2219
2220 if _, err := os.Stat(artifactsDir); err == nil {
2221 _ = filepath.WalkDir(artifactsDir, func(path string, d os.DirEntry, err error) error {
2222 if err != nil || d.IsDir() {
2223 return nil
2224 }
2225 rel, err := filepath.Rel(artifactsDir, path)
2226 if err != nil {
2227 return nil
2228 }
2229 if strings.HasPrefix(d.Name(), ".") {
2230 return nil
2231 }
2232 info, err := d.Info()
2233 if err != nil {
2234 return nil
2235 }
2236 artifacts = append(artifacts, artifactRow{
2237 Name: rel,
2238 Size: formatFileSize(info.Size()),
2239 ModTime: formatTimestamp(fmt.Sprintf("%d", info.ModTime().Unix())),
2240 })
2241 return nil
2242 })
2243 }
2244
2245 // Resolve overall job status
2246 jobStatus := "success"
2247 hasRunning := false
2248 for _, r := range rows {
2249 if r.Status == "running" {
2250 hasRunning = true
2251 break
2252 }
2253 if r.Status == "failed" {
2254 jobStatus = "failed"
2255 }
2256 }
2257 if hasRunning {
2258 jobStatus = "running"
2259 }
2260
2261 // HTML index
2262 tmpl, err := template.New("index").Funcs(template.FuncMap{
2263 "formatTimestamp": formatTimestamp,
2264 }).ParseFS(tmplFS, "tmpl/index.html")
2265 if err == nil {
2266 var buf strings.Builder
2267 if err := tmpl.ExecuteTemplate(&buf, "index.html", struct {
2268 Name string
2269 JobID string
2270 JobStatus string
2271 Rows []sessionRow
2272 Artifacts []artifactRow
2273 }{Name: name, JobID: jobID, JobStatus: jobStatus, Rows: rows, Artifacts: artifacts}); err == nil {
2274 htmlContent = buf.String()
2275 }
2276 }
2277
2278 // Plain-text index
2279 var buf strings.Builder
2280 statusIcon := map[string]string{"success": "\u2705", "failed": "\u274c", "running": "\u23f3"}
2281 icon := statusIcon[jobStatus]
2282 fmt.Fprintf(&buf, "Job: %s (%s) %s\n", name, jobID, icon)
2283 fmt.Fprintf(&buf, "Sessions: %d\n", len(rows))
2284 fmt.Fprintln(&buf, strings.Repeat("-", 70))
2285 for _, r := range rows {
2286 rIcon := statusIcon[r.Status]
2287 fmt.Fprintf(&buf, " %s %-20s exit: %-5s duration: %-8s ended: %s\n",
2288 rIcon, r.Short, r.ExitCode, r.Duration, r.Ended)
2289 }
2290 if len(artifacts) > 0 {
2291 fmt.Fprintln(&buf, "Artifacts:")
2292 for _, a := range artifacts {
2293 fmt.Fprintf(&buf, " %-30s %-10s %s\n", a.Name, a.Size, a.ModTime)
2294 }
2295 }
2296 fmt.Fprintln(&buf, strings.Repeat("-", 60))
2297 txtContent = buf.String()
2298
2299 return htmlContent, txtContent
2300}
2301
2302// fmtDuration formats the duration between two unix timestamp strings.
2303// Returns human-readable strings like "2m34s", "1.5s", or "—" if invalid.
2304func fmtDuration(created, ended string) string {
2305 if created == "" || ended == "" {
2306 return "—"
2307 }
2308 var c, e int64
2309 if _, err := fmt.Sscanf(created, "%d", &c); err != nil {
2310 return "—"
2311 }
2312 if _, err := fmt.Sscanf(ended, "%d", &e); err != nil {
2313 return "—"
2314 }
2315 duration := time.Duration(e-c) * time.Second
2316 if duration < 0 {
2317 return "—"
2318 }
2319 if duration >= time.Minute {
2320 return fmt.Sprintf("%dm%ds", duration/time.Minute, duration%time.Minute/time.Second)
2321 }
2322 secs := float64(duration) / float64(time.Second)
2323 if secs >= 10 {
2324 return fmt.Sprintf("%ds", int(secs))
2325 }
2326 return fmt.Sprintf("%.1fs", secs)
2327}
2328
2329func runLocal(cfg *Cfg, dest string) error {
2330 cwd, err := os.Getwd()
2331 if err != nil {
2332 return fmt.Errorf("get working directory: %w", err)
2333 }
2334
2335 picoPath := filepath.Join(cwd, "pico.sh")
2336 if _, err := os.Stat(picoPath); os.IsNotExist(err) {
2337 printMissingPicoHelp(cwd)
2338 return fmt.Errorf("no pico.sh found in %s", cwd)
2339 }
2340
2341 repoName := filepath.Base(cwd)
2342 jobID := fmt.Sprintf("%d", time.Now().Unix())
2343
2344 eventData := &Event{
2345 Type: "local",
2346 Name: repoName,
2347 JobID: jobID,
2348 Workspace: cwd,
2349 ArtifactDest: dest,
2350 }
2351
2352 // Process -e / --env flags
2353 for _, envPair := range cfg.EnvVars {
2354 parts := strings.SplitN(envPair, "=", 2)
2355 val := ""
2356 if len(parts) == 2 {
2357 val = parts[1]
2358 }
2359 key := parts[0]
2360 switch key {
2361 case "PICI_REPO":
2362 eventData.Name = val
2363 case "PICI_JOB":
2364 eventData.JobID = val
2365 case "PICI_EVENT":
2366 eventData.Type = val
2367 case "PICI_BRANCH":
2368 eventData.Branch = val
2369 case "PICI_COMMIT":
2370 eventData.Commit = val
2371 case "PICI_TAG":
2372 eventData.Tag = val
2373 }
2374 _ = os.Setenv(key, val)
2375 }
2376
2377 if dest != "" {
2378 eventData.ArtifactDest = strings.TrimSuffix(dest, "/") + "/" + eventData.JobID
2379 }
2380
2381 // Auto-detect git commit SHA and branch if not explicitly provided via -e
2382 if eventData.Commit == "" {
2383 eventData.Commit = detectGitCommit(cwd)
2384 }
2385 if eventData.Branch == "" {
2386 eventData.Branch = detectGitBranch(cwd)
2387 }
2388
2389 // Always block and output human format for local runs
2390 cfg.Wait = true
2391 cfg.HumanOutput = true
2392
2393 logger := cfg.Logger
2394 if logger == nil {
2395 logger = newLogger("ci", "info")
2396 }
2397 log := logger.With("repo", eventData.Name, "job_id", jobID, "mode", "local")
2398
2399 // Cancel any existing running job for this repo
2400 cancelRunningJobs(cfg, log, eventData.Name)
2401
2402 // Set up workspace
2403 var wk Workspace
2404 if cfg.InPlace {
2405 wk = &WorkspaceInPlace{
2406 Cfg: cfg,
2407 Logger: log,
2408 Source: cwd,
2409 }
2410 } else {
2411 wk = &WorkspaceRsync{
2412 Cfg: cfg,
2413 Logger: log,
2414 Source: cwd,
2415 }
2416 }
2417
2418 eng := &JobEngine{
2419 Logger: log,
2420 Cfg: cfg,
2421 Wk: wk,
2422 Ev: eventData,
2423 JobID: jobID,
2424 }
2425
2426 defer func() {
2427 if err := eng.Cleanup(); err != nil {
2428 log.Error("cleanup workspace", "err", err)
2429 }
2430 }()
2431
2432 log.Info("starting local job", "job_id", jobID)
2433 if err := eng.Setup(); err != nil {
2434 return fmt.Errorf("workspace setup: %w", err)
2435 }
2436 if cfg.InPlace {
2437 log.Debug("using in-place workspace directory", "dir", eng.Wk.GetDir())
2438 } else {
2439 log.Debug("syncing workspace to temp directory", "dir", eng.Wk.GetDir())
2440 }
2441 log.Debug("workspace directory", "dir", eng.Wk.GetDir())
2442
2443 // Store event.json in artifact dir
2444 eventDir := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID)
2445 artifactsDir := filepath.Join(eventDir, "artifacts")
2446 if err := os.MkdirAll(artifactsDir, 0755); err != nil {
2447 log.Error("create artifacts dir", "err", err)
2448 } else {
2449 eventBytes, _ := json.Marshal(eventData)
2450 _ = os.WriteFile(filepath.Join(artifactsDir, "event.json"), eventBytes, 0644)
2451 }
2452
2453 // Write attestation.json
2454 hostname, _ := os.Hostname()
2455 attestation := map[string]interface{}{
2456 "runner": map[string]string{
2457 "hostname": hostname,
2458 "os": runtimeOS(),
2459 "arch": runtimeArch(),
2460 },
2461 "provenance": map[string]string{
2462 "repo": eventData.Name,
2463 "branch": eventData.Branch,
2464 "commit": eventData.Commit,
2465 },
2466 "workspace_checksum": eng.Wk.Checksum(),
2467 }
2468 attestationBytes, _ := json.Marshal(attestation)
2469 _ = os.WriteFile(filepath.Join(artifactsDir, "attestation.json"), attestationBytes, 0644)
2470
2471 manifest, err := eng.FindManifest()
2472 if err != nil {
2473 return err
2474 }
2475
2476 log.Debug("launching sessions")
2477 if err := eng.Run(manifest); err != nil {
2478 return fmt.Errorf("run: %w", err)
2479 }
2480
2481 // Wait for completion & print live progress
2482 var waitErr error
2483 if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
2484 waitErr = err
2485 }
2486
2487 // Generate and stage full HTML/txt artifacts and index
2488 if err := stageLocalArtifacts(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
2489 log.Error("stage local artifacts", "err", err)
2490 }
2491
2492 indexFile := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID, "index.html")
2493 fmt.Fprintf(os.Stdout, "📊 local html report: file://%s\n", indexFile) //nolint:errcheck
2494
2495 // Sync to destination if specified
2496 if dest != "" {
2497 fmt.Fprintf(os.Stdout, "🔄 rsyncing artifacts to %s...\n", eventData.ArtifactDest) //nolint:errcheck
2498 if err := syncJobArtifacts(cfg, eventData.Name, eventData.JobID, log); err != nil {
2499 return fmt.Errorf("sync artifacts: %w", err)
2500 }
2501 fmt.Fprintf(os.Stdout, "✅ artifacts rsynced to %s\n", eventData.ArtifactDest) //nolint:errcheck
2502 }
2503
2504 // Write published sentinel for local run (`pici`)
2505 domain := getDomain(eventData.Type)
2506 prefix := fmt.Sprintf("%s.%s.%s.", domain, eventData.Name, jobID)
2507 if listOutput, err := exec.Command("zmx", "list").CombinedOutput(); err == nil {
2508 sessions := parseZMXList(string(listOutput))
2509 var jobSessions []SessionInfo
2510 for _, s := range sessions {
2511 if strings.HasPrefix(s.Name, prefix) {
2512 s.Short = cleanSessionShort(s.Name, prefix, eventData.Name, jobID)
2513 jobSessions = append(jobSessions, s)
2514 }
2515 }
2516 exitCode, status := resolveJobExitCode(jobSessions)
2517 _ = writePublishedSentinel(cfg.ArtifactDir, eventData.Name, jobID, status, exitCode)
2518 }
2519
2520 if waitErr != nil {
2521 return waitErr
2522 }
2523
2524 return nil
2525}
2526
2527func writePublishedSentinel(artifactDir, repoName, jobID, status string, exitCode int) error {
2528 sentinel := filepath.Join(artifactDir, repoName, jobID, "artifacts", "published.json")
2529 published := map[string]interface{}{
2530 "status": status,
2531 "exit_code": exitCode,
2532 "job_id": jobID,
2533 "finished_at": time.Now().UTC().Format(time.RFC3339),
2534 }
2535 publishedJSON, err := json.Marshal(published)
2536 if err != nil {
2537 return err
2538 }
2539 if err := os.MkdirAll(filepath.Dir(sentinel), 0755); err != nil {
2540 return err
2541 }
2542 return os.WriteFile(sentinel, publishedJSON, 0644)
2543}
2544
2545func stageLocalArtifacts(cfg *Cfg, log *slog.Logger, repoName, jobID, eventType string) error {
2546 domain := getDomain(eventType)
2547 prefix := fmt.Sprintf("%s.%s.%s.", domain, repoName, jobID)
2548
2549 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
2550 if err != nil {
2551 return fmt.Errorf("zmx list: %w", err)
2552 }
2553 sessions := parseZMXList(string(listOutput))
2554
2555 var jobSessions []SessionInfo
2556 for _, s := range sessions {
2557 if strings.HasPrefix(s.Name, prefix) {
2558 s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
2559 jobSessions = append(jobSessions, s)
2560 }
2561 }
2562
2563 for _, s := range jobSessions {
2564 sessionStatus := "running"
2565 sessionDuration := fmtDuration(s.Created, fmt.Sprintf("%d", time.Now().Unix()))
2566 sessionExitCode := ""
2567 if s.Ended != "" {
2568 sessionDuration = fmtDuration(s.Created, s.Ended)
2569 if s.ExitCode == "0" {
2570 sessionStatus = "success"
2571 sessionExitCode = "0"
2572 } else {
2573 sessionStatus = "failed"
2574 sessionExitCode = s.ExitCode
2575 }
2576 }
2577
2578 html, err := fetchHistoryHTML(s.Name, repoName, jobID, sessionStatus, sessionDuration, sessionExitCode)
2579 if err == nil {
2580 _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, html, ".html")
2581 }
2582 plain, err := fetchHistoryPlain(s.Name)
2583 if err == nil {
2584 _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, plain, ".txt")
2585 }
2586 }
2587
2588 indexHTML, indexTXT := generateJobIndex(cfg.ArtifactDir, repoName, jobID, jobSessions)
2589 _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexHTML, ".html")
2590 _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexTXT, ".txt")
2591 if styles, err := loadStyles(); err == nil {
2592 _ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "styles", styles, ".css")
2593 }
2594
2595 return nil
2596}
2597
2598func detectGitCommit(dir string) string {
2599 cmd := exec.Command("git", "-C", dir, "rev-parse", "HEAD")
2600 out, err := cmd.Output()
2601 if err != nil {
2602 return ""
2603 }
2604 return strings.TrimSpace(string(out))
2605}
2606
2607func detectGitBranch(dir string) string {
2608 cmd := exec.Command("git", "-C", dir, "rev-parse", "--abbrev-ref", "HEAD")
2609 out, err := cmd.Output()
2610 if err != nil || strings.TrimSpace(string(out)) == "HEAD" {
2611 return ""
2612 }
2613 return strings.TrimSpace(string(out))
2614}