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