pici

feat: add local runner base command with isolated execution and html logging rss

created pr with 132.1 on 2026-08-08T17:48:30Z · by c8ef7d19

cmds

checkout latest patchset:

ssh pr.pico.sh print 132 | git am -3

checkout any patchset in a patch request:

ssh pr.pico.sh print 132.[rev] | git am -3

add changes to patch request:

git format-patch main --stdout | ssh pr.pico.sh pr add 132

Patchset `132.1` on 2026-08-08T17:48:30Z · commit 0224564

→ feat: add local runner base command with isolated execution and html logging

Eric Bower 2026-08-08T15:54:27Z

feat: add local runner base command with isolated execution and html logging

Eric Bower 2026-08-08T15:54:27Z

- Default `pici [destination]` command copies CWD to /tmp for isolated execution
- Streamline CLI flags with -e/--env for key-value environment overrides
- Isolate local sessions with `local.` prefix so `pici monitor` daemons ignore local dev runs
- Auto-detect git commit SHA and branch for attestation.json and event metadata
- Rsync build artifacts directly into a `<job_id>` subfolder at target destination
- Add onboarding guide when pico.sh is missing and root `pici help` overview

Semantic diff summary

12 added, 15 modified, 1 signature changed, 0 removed across 2 analyzed files

main.go

method_declaration String added
method_declaration Set added
type_declaration envList added
chunk lines 72-78 modified
function_declaration NewCfg modified
function_declaration printMainHelp added
function_declaration printMissingPicoHelp added
function_declaration main modified
method_declaration Cleanup modified
method_declaration Run modified
function_declaration Run modified
function_declaration eventHandler modified
chunk lines 785-799 modified
function_declaration waitAndReport signature changed
function_declaration cleanSessionShort modified
function_declaration runMonitor modified
function_declaration syncJobArtifacts modified
function_declaration runGC modified
function_declaration detectGitCommit added
function_declaration detectGitBranch added
function_declaration runLocal added
function_declaration stageLocalArtifacts added

main_test.go

function_declaration TestE2E_RunnerWithZMXSessions modified
chunk lines 46-56 modified
chunk lines 114-120 modified
function_declaration TestDetectGitCommit added
function_declaration TestRunLocal_MissingPico added
function_declaration TestRunLocal_EnvOverrides added

+394 -32 main.go #

......@@ -45,6 +45,17 @@ func defaultWorkspaceFactory(cfg *Cfg, logger *slog.Logger, source string) Works
	}
}

type envList []string

func (e *envList) String() string {
	return strings.Join(*e, ", ")
}

func (e *envList) Set(value string) error {
	*e = append(*e, value)
	return nil
}

type Cfg struct {
	Logger              *slog.Logger
	Ctx                 context.Context

......@@ -61,6 +72,7 @@ type Cfg struct {
	IncludeRunning      bool      // emit running status updates in addition to terminal
	HumanOutput         bool      // human-readable output instead of JSONL / slog
	Wait                bool      // block until job completes, print history and summary
	EnvVars             envList   // custom environment variables passed via -e / -env
}

type Event struct {

......@@ -80,6 +92,7 @@ func NewCfg() (*Cfg, string, bool) {
	var monitorInterval time.Duration
	var gcInterval time.Duration
	var logLevel string
	var envVars envList
	flag.StringVar(&keyLoc, "pk", "", "ssh private key used to authenticate with pico services")
	flag.StringVar(&certLoc, "ck", "", "ssh certificate public key used to authenticate with pico services (only required if using ssh certificates)")
	flag.StringVar(&artifactDir, "artifact-dir", "/tmp/pici-artifacts", "local directory to stage artifacts")

......@@ -87,6 +100,8 @@ func NewCfg() (*Cfg, string, bool) {
	flag.DurationVar(&monitorInterval, "monitor-interval", 5*time.Second, "interval for monitoring zmx sessions")
	flag.DurationVar(&gcInterval, "gc-interval", 10*time.Minute, "interval for garbage collection in monitor (0 to disable)")
	flag.StringVar(&logLevel, "log-level", "info", "log level: debug, info, warn, error")
	flag.Var(&envVars, "e", "environment variable in KEY=VAL format (can be specified multiple times)")
	flag.Var(&envVars, "env", "environment variable in KEY=VAL format (can be specified multiple times)")
	var includeRunning bool
	var human bool
	var wait bool

......@@ -118,20 +133,30 @@ func NewCfg() (*Cfg, string, bool) {
		IncludeRunning:      includeRunning,
		HumanOutput:         human,
		Wait:                wait,
		EnvVars:             envVars,
	}, cmd, wantHelp
}

func isKnownSubcommand(s string) bool {
	switch s {
	case "runner", "monitor", "cancel", "gc", "status", "orca", "run", "help":
		return true
	default:
		return false
	}
}

// splitCommand separates the first non-flag argument (the subcommand) from
// the rest of the flags, so "runner --wait" becomes flags=["--wait"], cmd="runner".
// It strips --help/help so we can print custom help per subcommand.
func splitCommand(args []string) (flags []string, cmd string, wantHelp bool) {
	flags = make([]string, 0, len(args))
	for _, arg := range args {
		if arg == "--help" || arg == "help" {
		if arg == "--help" || arg == "help" || arg == "-h" {
			wantHelp = true
			continue
		}
		if cmd == "" && arg != "" && !strings.HasPrefix(arg, "-") {
		if cmd == "" && isKnownSubcommand(arg) {
			cmd = arg
		} else {
			flags = append(flags, arg)

......@@ -160,17 +185,67 @@ func newLogger(space string, levelStr string) *slog.Logger {
	})).With("service", space)
}

func printMainHelp() {
	fmt.Println(`pici — minimal parallel CI runner & monitor powered by zmx

LOCAL DEVELOPER USAGE
  pici [destination] [flags]                  Run ./pico.sh locally in /tmp & render HTML logs
  pici pgs.sh:/my-site                        Run locally and rsync HTML logs to destination
  pici run [destination]                      Explicit alias for local run

DAEMON & SERVICE COMMANDS
  pici runner                                 Execute CI job from event JSON payload (stdin/flag)
  pici monitor                                Poll ci.* zmx sessions & stage/sync HTML artifacts
  pici cancel                                 Cancel active running jobs for a repository
  pici gc                                     Clean up stale/finished zmx sessions & artifacts

SUBCOMMAND HELP
  pici <command> --help                       Show detailed help for a specific command (e.g. pici runner --help)

FLAGS
  -e, -env <KEY=VAL>                          Set or override environment variable for pico.sh
  -pk <path>                                  SSH private key
  -ck <path>                                  SSH public certificate key
  -artifact-dir <path>                        Artifact staging directory (default: /tmp/pici-artifacts)
  -log-level <level>                          Log level: debug, info, warn, error`)
}

func printMissingPicoHelp(cwd string) {
	fmt.Printf(`❌ Error: no pico.sh found in %s

To run local CI tasks, create a pico.sh script in your project root:

  #!/usr/bin/env bash
  set -euo pipefail

  # Run parallel steps using zmx
  zmx run lint -d <your lint command>
  zmx run test -d <your test command>

  # Wait for all steps to finish
  zmx wait "*"
  printf "\x1b[32msuccess!\x1b[0m\n"

For full command documentation and daemon options, run:
  pici help

`, cwd)
}

func main() {
	cfg, cmd, wantHelp := NewCfg()

	cfg.Logger.Debug("setting up ci", "cfg", cfg)
	cfg.Logger.Debug("running cmd", "cmd", cmd)

	if wantHelp && (cmd == "runner" || cmd == "monitor") {
		if cmd == "runner" {
	if wantHelp {
		switch cmd {
		case "runner":
			printRunnerHelp()
		} else {
		case "monitor":
			printMonitorHelp()
		default:
			printMainHelp()
		}
		return
	}

......@@ -204,9 +279,23 @@ func main() {
		cfg.Logger.Debug("starting status updater")
	case "orca":
		cfg.Logger.Debug("starting orchestrator")
	case "help":
		printMainHelp()
	case "run", "":
		dest := ""
		if flag.NArg() > 0 {
			dest = flag.Arg(0)
		}
		if err := runLocal(cfg, dest); err != nil {
			cfg.Logger.Error("local run failed", "err", err)
			os.Exit(1)
		}
	default:
		cfg.Logger.Error("must provide command: runner, cancel, gc, monitor, status, or orca")
		os.Exit(1)
		dest := cmd
		if err := runLocal(cfg, dest); err != nil {
			cfg.Logger.Error("local run failed", "err", err)
			os.Exit(1)
		}
	}
}

......@@ -362,7 +451,9 @@ func (w *WorkspaceRsync) Setup() error {
}

func (w *WorkspaceRsync) Cleanup() error {
	// return os.RemoveAll(w.Dest)
	if w.Dest != "" {
		return os.RemoveAll(w.Dest)
	}
	return nil
}

......@@ -522,9 +613,13 @@ func (eng *JobEngine) Setup() error {
}

func (eng *JobEngine) Run(manifest string) error {
	prefix := fmt.Sprintf("ci.%s.%s.", eng.Ev.Name, eng.JobID)
	domain := "ci"
	if eng.Ev != nil && eng.Ev.Type == "local" {
		domain = "local"
	}
	prefix := fmt.Sprintf("%s.%s.%s.", domain, eng.Ev.Name, eng.JobID)
	// Child sessions use ".step." sub-prefix so zmx wait "*" inside pico.sh
	// matches ci.<name>.<jobID>.step.* but NOT ci.<name>.<jobID>.runner.
	// matches <domain>.<name>.<jobID>.step.* but NOT <domain>.<name>.<jobID>.runner.
	// This avoids a deadlock where the runner waits for itself.
	childPrefix := prefix + "step."

......@@ -551,9 +646,8 @@ func (eng *JobEngine) Run(manifest string) error {
	if eng.Ev.Tag != "" {
		cmdEnv = append(cmdEnv, fmt.Sprintf("PICI_TAG=%s", eng.Ev.Tag))
	}

	zmxPrefixStr := fmt.Sprintf("ZMX_SESSION_PREFIX=%s", childPrefix)
	cmd := exec.Command("zmx", "run", runnerName, "-d", zmxPrefixStr, "bash", manifest)
	bashCmd := fmt.Sprintf("export ZMX_SESSION_PREFIX=%q; exec bash %q", childPrefix, manifest)
	cmd := exec.Command("zmx", "run", runnerName, "-d", "bash", "-c", bashCmd)
	cmd.Env = cmdEnv
	cmd.Dir = eng.Wk.GetDir()

......@@ -610,14 +704,18 @@ func eventHandler(cfg *Cfg, eventData *Event) error {
		Ev:     eventData,
		JobID:  jobID,
	}
	var runErr error
	defer func() {
		if err := eng.Cleanup(); err != nil {
			cfg.Logger.Error("engine cleanup", "err", err)
		if runErr != nil || cfg.Wait {
			if err := eng.Cleanup(); err != nil {
				cfg.Logger.Error("engine cleanup", "err", err)
			}
		}
	}()

	fmt.Fprintf(os.Stdout, "📦 syncing workspace %s\n", eventData.Workspace) //nolint:errcheck
	if err := eng.Setup(); err != nil {
		runErr = err
		return fmt.Errorf("setup: %w", err)
	}
	fmt.Fprintf(os.Stdout, "✅ workspace ready %s\n", eng.Wk.GetDir()) //nolint:errcheck

......@@ -654,6 +752,7 @@ func eventHandler(cfg *Cfg, eventData *Event) error {

	manifest, err := eng.FindManifest()
	if err != nil {
		runErr = err
		fmt.Fprintf(os.Stdout, "❌ %s\n\n", err) //nolint:errcheck
		//nolint:errcheck
		fmt.Fprint(os.Stdout, `Create a pico.sh script in your workspace root:

......@@ -686,13 +785,15 @@ See: https://github.com/picosh/pici

	fmt.Fprint(os.Stdout, "🏃 launching sessions...\n") //nolint:errcheck
	if err := eng.Run(manifest); err != nil {
		runErr = err
		return fmt.Errorf("run: %w", err)
	}

	fmt.Fprintln(os.Stdout, "✅ job launched") //nolint:errcheck

	if cfg.Wait {
		if err := waitAndReport(cfg, log, eventData.Name, jobID); err != nil {
		if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
			runErr = err
			return fmt.Errorf("wait: %w", err)
		}
		return nil

......@@ -707,9 +808,17 @@ See: https://github.com/picosh/pici

// waitAndReport polls the job's sessions until all complete, prints live
// progress to stdout, then dumps session history and a final summary.
func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID string) error {
	prefix := "ci." + name + "." + jobID + "."
	ticker := time.NewTicker(cfg.MonitorInterval)
func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) error {
	domain := "ci"
	if eventType == "local" {
		domain = "local"
	}
	prefix := domain + "." + name + "." + jobID + "."
	interval := cfg.MonitorInterval
	if interval <= 0 {
		interval = 5 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	// Handle ^C gracefully

......@@ -726,9 +835,14 @@ func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID string) error {
	var sessionOrder []string // track insertion order for deterministic output
	var liveLines []string    // last set of status lines printed (for overwrite)

	var done <-chan struct{}
	if cfg.Ctx != nil {
		done = cfg.Ctx.Done()
	}

	for {
		select {
		case <-cfg.Ctx.Done():
		case <-done:
			return cfg.Ctx.Err()
		case <-sigCh:
			fmt.Fprintln(os.Stdout, "\n⏹ cancelled") //nolint:errcheck

......@@ -876,9 +990,11 @@ func cleanSessionShort(name, prefix, repoName, jobID string) string {
	short := strings.TrimPrefix(name, prefix)
	// Strip "step." prefix added by child sessions
	short = strings.TrimPrefix(short, "step.")
	// Strip nested full prefix (e.g. runner named ci.name.jobID.runner)
	nested := "ci." + repoName + "." + jobID + "."
	short = strings.TrimPrefix(short, nested)
	// Strip nested full prefix (e.g. runner named ci.name.jobID.runner or local.name.jobID.runner)
	nestedCI := "ci." + repoName + "." + jobID + "."
	short = strings.TrimPrefix(short, nestedCI)
	nestedLocal := "local." + repoName + "." + jobID + "."
	short = strings.TrimPrefix(short, nestedLocal)
	return short
}

......@@ -917,10 +1033,11 @@ func runMonitor(cfg *Cfg) error {
	defer ticker.Stop()

	// Optional GC ticker — runs garbage collection on a separate interval.
	var gcTicker *time.Ticker
	var gcChan <-chan time.Time
	if cfg.GCInterval > 0 {
		gcTicker = time.NewTicker(cfg.GCInterval)
		gcTicker := time.NewTicker(cfg.GCInterval)
		defer gcTicker.Stop()
		gcChan = gcTicker.C
	}

	// Track per-job display state across ticks (for human output)

......@@ -938,7 +1055,7 @@ func runMonitor(cfg *Cfg) error {
			if err := monitorTick(cfg, log, output, jobStates); err != nil {
				log.Error("monitor tick", "err", err)
			}
		case <-gcTicker.C:
		case <-gcChan:
			log.Debug("running periodic garbage collection")
			if err := runGC(cfg); err != nil {
				log.Error("periodic gc", "err", err)

......@@ -1560,21 +1677,21 @@ func syncJobArtifacts(cfg *Cfg, repoName, jobID string, log *slog.Logger) error
		}
		sshArgs = fmt.Sprintf("-F ~/.ssh/config -i %s%s", cfg.KeyLocation, certFile)
	}
	// Append "/" so rsync copies into the destination directory,
	// not as a subdirectory named after the source.
	// Source is jobDir (without trailing slash) so rsync creates
	// the jobID subdirectory inside the destination folder.
	dest := event.ArtifactDest
	if !strings.HasSuffix(dest, "/") {
		dest += "/"
	}
	var cmd *exec.Cmd
	if sshArgs != "" {
		cmd = exec.Command("rsync", "-e", sshArgs, "-rv", jobDir+"/", dest)
		cmd = exec.Command("rsync", "-e", sshArgs, "-rv", jobDir, dest)
	} else {
		cmd = exec.Command("rsync", "-rv", jobDir+"/", dest)
		cmd = exec.Command("rsync", "-rv", jobDir, dest)
	}
	rsyncCmd := fmt.Sprintf("rsync %s %s %s",
		strings.TrimLeft(cmd.Args[1], "-"),
		jobDir+"/", dest)
		jobDir, dest)
	log.Info("rsync", "cmd", rsyncCmd)
	return runCmd(cmd, log)
}

......@@ -1730,7 +1847,7 @@ func runGC(cfg *Cfg) error {

	var toKill []string
	for _, s := range sessions {
		if !strings.HasPrefix(s.Name, "ci.") {
		if !strings.HasPrefix(s.Name, "ci.") && !strings.HasPrefix(s.Name, "local.") {
			continue
		}

......@@ -1951,3 +2068,248 @@ func fmtDuration(created, ended string) string {
	}
	return fmt.Sprintf("%.1fs", secs)
}

func runLocal(cfg *Cfg, dest string) error {
	cwd, err := os.Getwd()
	if err != nil {
		return fmt.Errorf("get working directory: %w", err)
	}

	picoPath := filepath.Join(cwd, "pico.sh")
	if _, err := os.Stat(picoPath); os.IsNotExist(err) {
		printMissingPicoHelp(cwd)
		return fmt.Errorf("no pico.sh found in %s", cwd)
	}

	repoName := filepath.Base(cwd)
	jobID := fmt.Sprintf("local-%d", time.Now().Unix())

	eventData := &Event{
		Type:         "local",
		Name:         repoName,
		JobID:        jobID,
		Workspace:    cwd,
		ArtifactDest: dest,
	}

	// Process -e / --env flags
	for _, envPair := range cfg.EnvVars {
		parts := strings.SplitN(envPair, "=", 2)
		val := ""
		if len(parts) == 2 {
			val = parts[1]
		}
		key := parts[0]
		switch key {
		case "PICI_REPO":
			eventData.Name = val
		case "PICI_JOB":
			eventData.JobID = val
		case "PICI_EVENT":
			eventData.Type = val
		case "PICI_BRANCH":
			eventData.Branch = val
		case "PICI_COMMIT":
			eventData.Commit = val
		case "PICI_TAG":
			eventData.Tag = val
		}
		_ = os.Setenv(key, val)
	}

	// Auto-detect git commit SHA and branch if not explicitly provided via -e
	if eventData.Commit == "" {
		eventData.Commit = detectGitCommit(cwd)
	}
	if eventData.Branch == "" {
		eventData.Branch = detectGitBranch(cwd)
	}

	// Always block and output human format for local runs
	cfg.Wait = true
	cfg.HumanOutput = true

	logger := cfg.Logger
	if logger == nil {
		logger = newLogger("ci", "info")
	}
	log := logger.With("repo", eventData.Name, "job_id", jobID, "mode", "local")

	// Set up workspace in temp directory
	wk := &WorkspaceRsync{
		Cfg:    cfg,
		Logger: log,
		Source: cwd,
	}

	eng := &JobEngine{
		Logger: log,
		Cfg:    cfg,
		Wk:     wk,
		Ev:     eventData,
		JobID:  jobID,
	}

	defer func() {
		if err := eng.Cleanup(); err != nil {
			log.Error("cleanup workspace", "err", err)
		}
	}()

	fmt.Fprintf(os.Stdout, "🚀 starting local job local.%s.%s\n", eventData.Name, jobID) //nolint:errcheck
	fmt.Fprintln(os.Stdout, "📦 syncing workspace to temp directory...")                 //nolint:errcheck
	if err := eng.Setup(); err != nil {
		return fmt.Errorf("workspace setup: %w", err)
	}
	log.Debug("workspace directory", "dir", eng.Wk.GetDir())

	// Store event.json in artifact dir
	eventDir := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID)
	artifactsDir := filepath.Join(eventDir, "artifacts")
	if err := os.MkdirAll(artifactsDir, 0755); err != nil {
		log.Error("create artifacts dir", "err", err)
	} else {
		eventBytes, _ := json.Marshal(eventData)
		_ = os.WriteFile(filepath.Join(artifactsDir, "event.json"), eventBytes, 0644)
	}

	// Write attestation.json
	hostname, _ := os.Hostname()
	attestation := map[string]interface{}{
		"runner": map[string]string{
			"hostname": hostname,
			"os":       runtimeOS(),
			"arch":     runtimeArch(),
		},
		"provenance": map[string]string{
			"repo":   eventData.Name,
			"branch": eventData.Branch,
			"commit": eventData.Commit,
		},
		"workspace_checksum": eng.Wk.Checksum(),
	}
	attestationBytes, _ := json.Marshal(attestation)
	_ = os.WriteFile(filepath.Join(artifactsDir, "attestation.json"), attestationBytes, 0644)

	manifest, err := eng.FindManifest()
	if err != nil {
		return err
	}

	fmt.Fprintln(os.Stdout, "🏃 launching sessions...") //nolint:errcheck
	if err := eng.Run(manifest); err != nil {
		return fmt.Errorf("run: %w", err)
	}

	// Wait for completion & print live progress
	if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
		return fmt.Errorf("wait: %w", err)
	}

	// Generate and stage full HTML/txt artifacts and index
	if err := stageLocalArtifacts(cfg, log, eventData.Name, jobID); err != nil {
		log.Error("stage local artifacts", "err", err)
	}

	indexFile := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID, "index.html")
	fmt.Fprintf(os.Stdout, "📊 local html report: file://%s\n", indexFile) //nolint:errcheck

	// Sync to destination if specified
	if dest != "" {
		targetDest := strings.TrimSuffix(dest, "/") + "/" + jobID
		fmt.Fprintf(os.Stdout, "🔄 rsyncing artifacts to %s...\n", dest) //nolint:errcheck
		if err := syncJobArtifacts(cfg, eventData.Name, jobID, log); err != nil {
			return fmt.Errorf("sync artifacts: %w", err)
		}
		fmt.Fprintf(os.Stdout, "✅ artifacts rsynced to %s\n", targetDest) //nolint:errcheck
	}

	return nil
}

func stageLocalArtifacts(cfg *Cfg, log *slog.Logger, repoName, jobID string) error {
	listOutput, err := exec.Command("zmx", "list").CombinedOutput()
	if err != nil {
		return fmt.Errorf("zmx list: %w", err)
	}
	sessions := parseZMXList(string(listOutput))
	var localSessions []SessionInfo
	for _, s := range sessions {
		if strings.HasPrefix(s.Name, "local.") {
			localSessions = append(localSessions, s)
		}
	}
	prefix := fmt.Sprintf("local.%s.%s.", repoName, jobID)

	var jobSessions []SessionInfo
	for _, s := range localSessions {
		if strings.HasPrefix(s.Name, prefix) {
			s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
			jobSessions = append(jobSessions, s)
		}
	}

	for _, s := range jobSessions {
		sessionStatus := "running"
		sessionDuration := fmtDuration(s.Created, fmt.Sprintf("%d", time.Now().Unix()))
		sessionExitCode := ""
		if s.Ended != "" {
			sessionDuration = fmtDuration(s.Created, s.Ended)
			if s.ExitCode == "0" {
				sessionStatus = "success"
				sessionExitCode = "0"
			} else {
				sessionStatus = "failed"
				sessionExitCode = s.ExitCode
			}
		}

		html, err := fetchHistoryHTML(s.Name, repoName, jobID, sessionStatus, sessionDuration, sessionExitCode)
		if err == nil {
			_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, html, ".html")
		}
		plain, err := fetchHistoryPlain(s.Name)
		if err == nil {
			_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, plain, ".txt")
		}
	}

	// Write published sentinel
	exitCode, status := resolveJobExitCode(jobSessions)
	sentinel := filepath.Join(cfg.ArtifactDir, repoName, jobID, "artifacts", "published.json")
	published := map[string]interface{}{
		"status":      status,
		"exit_code":   exitCode,
		"job_id":      jobID,
		"finished_at": time.Now().UTC().Format(time.RFC3339),
	}
	publishedJSON, _ := json.Marshal(published)
	_ = os.WriteFile(sentinel, publishedJSON, 0644)

	indexHTML, indexTXT := generateJobIndex(cfg.ArtifactDir, repoName, jobID, jobSessions)
	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexHTML, ".html")
	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexTXT, ".txt")
	if styles, err := loadStyles(); err == nil {
		_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "styles", styles, ".css")
	}

	return nil
}

func detectGitCommit(dir string) string {
	cmd := exec.Command("git", "-C", dir, "rev-parse", "HEAD")
	out, err := cmd.Output()
	if err != nil {
		return ""
	}
	return strings.TrimSpace(string(out))
}

func detectGitBranch(dir string) string {
	cmd := exec.Command("git", "-C", dir, "rev-parse", "--abbrev-ref", "HEAD")
	out, err := cmd.Output()
	if err != nil || strings.TrimSpace(string(out)) == "HEAD" {
		return ""
	}
	return strings.TrimSpace(string(out))
}

+62 -4 main_test.go #

......@@ -25,17 +25,20 @@ func TestE2E_RunnerWithZMXSessions(t *testing.T) {
	if testing.Short() {
		t.Skip("skip integration test")
	}
	if _, err := exec.LookPath("zmx"); err != nil {
	zmxPath, err := exec.LookPath("zmx")
	if err != nil {
		t.Skip("zmx not found, skipping integration test")
	}
	zmxDir := filepath.Dir(zmxPath)

	// 1. Create workspace with pico.sh that spawns zmx sessions
	workspaceDir := t.TempDir()
	picoSh := `#!/usr/bin/env bash
	picoSh := fmt.Sprintf(`#!/usr/bin/env bash
set -e
export PATH="%s:$PATH"
zmx run step1 echo "hello from step1"
zmx run step2 echo "hello from step2"
`
`, zmxDir)
	if err := os.WriteFile(filepath.Join(workspaceDir, "pico.sh"), []byte(picoSh), 0755); err != nil {
		t.Fatalf("write pico.sh: %v", err)
	}

......@@ -43,9 +46,11 @@ zmx run step2 echo "hello from step2"
	// 2. Create config
	artifactDir := t.TempDir()
	ctx, cancel := context.WithCancel(context.Background())
	testJobID := fmt.Sprintf("j-%d", time.Now().UnixNano()%100000000)
	event := Event{
		Type:      "build",
		Name:      "test-repo",
		JobID:     testJobID,
		Workspace: workspaceDir,
	}
	eventJSON, _ := json.Marshal(event)

......@@ -109,7 +114,7 @@ zmx run step2 echo "hello from step2"
				}

				// Ignore statuses from unrelated jobs (e.g., leftover sessions from previous tests)
				if p.Name != "test-repo" {
				if p.Name != "test-repo" || p.JobID != testJobID {
					continue
				}

......@@ -690,3 +695,56 @@ func TestResolveJobExitCode(t *testing.T) {
		})
	}
}

func TestRunLocal_MissingPico(t *testing.T) {
	tempDir := t.TempDir()
	origWd, _ := os.Getwd()
	defer func() { _ = os.Chdir(origWd) }()
	_ = os.Chdir(tempDir)

	cfg := &Cfg{
		ArtifactDir: t.TempDir(),
	}
	err := runLocal(cfg, "")
	if err == nil {
		t.Fatal("expected error when pico.sh is missing")
	}
}

func TestRunLocal_EnvOverrides(t *testing.T) {
	tempDir := t.TempDir()
	origWd, _ := os.Getwd()
	defer func() { _ = os.Chdir(origWd) }()
	_ = os.Chdir(tempDir)

	picoContent := `#!/usr/bin/env bash
echo "hello from pico"
`
	if err := os.WriteFile("pico.sh", []byte(picoContent), 0755); err != nil {
		t.Fatal(err)
	}

	cfg := &Cfg{
		ArtifactDir: t.TempDir(),
		EnvVars:     envList{"PICI_REPO=custom-repo", "CUSTOM_VAR=hello"},
	}

	_ = runLocal(cfg, "")

	if os.Getenv("CUSTOM_VAR") != "hello" {
		t.Errorf("expected CUSTOM_VAR to be hello, got %q", os.Getenv("CUSTOM_VAR"))
	}
}

func TestDetectGitCommit(t *testing.T) {
	cwd, err := os.Getwd()
	if err != nil {
		t.Fatal(err)
	}
	commit := detectGitCommit(cwd)
	if commit == "" {
		t.Log("git commit sha not detected (not a git repository or git unavailable)")
	} else {
		t.Logf("detected git commit sha: %s", commit)
	}
}

...	...		@@ -118,20 +133,30 @@ func NewCfg() (*Cfg, string, bool) {
118	133		IncludeRunning: includeRunning,
119	134		HumanOutput: human,
120	135		Wait: wait,
	136	+	EnvVars: envVars,
121	137		}, cmd, wantHelp
122	138		}
123	139
	140	+	func isKnownSubcommand(s string) bool {
	141	+	switch s {
	142	+	case "runner", "monitor", "cancel", "gc", "status", "orca", "run", "help":
	143	+	return true
	144	+	default:
	145	+	return false
	146	+	}
	147	+	}
	148	+
124	149		// splitCommand separates the first non-flag argument (the subcommand) from
125	150		// the rest of the flags, so "runner --wait" becomes flags=["--wait"], cmd="runner".
126	151		// It strips --help/help so we can print custom help per subcommand.
127	152		func splitCommand(args []string) (flags []string, cmd string, wantHelp bool) {
128	153		flags = make([]string, 0, len(args))
129	154		for _, arg := range args {
130		-	if arg == "--help" \|\| arg == "help" {
	155	+	if arg == "--help" \|\| arg == "help" \|\| arg == "-h" {
131	156		wantHelp = true
132	157		continue
133	158		}
134		-	if cmd == "" && arg != "" && !strings.HasPrefix(arg, "-") {
	159	+	if cmd == "" && isKnownSubcommand(arg) {
135	160		cmd = arg
136	161		} else {
137	162		flags = append(flags, arg)

...	...		@@ -160,17 +185,67 @@ func newLogger(space string, levelStr string) *slog.Logger {
160	185		})).With("service", space)
161	186		}
162	187
	188	+	func printMainHelp() {
	189	+	fmt.Println(`pici — minimal parallel CI runner & monitor powered by zmx
	190	+
	191	+	LOCAL DEVELOPER USAGE
	192	+	pici [destination] [flags] Run ./pico.sh locally in /tmp & render HTML logs
	193	+	pici pgs.sh:/my-site Run locally and rsync HTML logs to destination
	194	+	pici run [destination] Explicit alias for local run
	195	+
	196	+	DAEMON & SERVICE COMMANDS
	197	+	pici runner Execute CI job from event JSON payload (stdin/flag)
	198	+	pici monitor Poll ci.* zmx sessions & stage/sync HTML artifacts
	199	+	pici cancel Cancel active running jobs for a repository
	200	+	pici gc Clean up stale/finished zmx sessions & artifacts
	201	+
	202	+	SUBCOMMAND HELP
	203	+	pici <command> --help Show detailed help for a specific command (e.g. pici runner --help)
	204	+
	205	+	FLAGS
	206	+	-e, -env <KEY=VAL> Set or override environment variable for pico.sh
	207	+	-pk <path> SSH private key
	208	+	-ck <path> SSH public certificate key
	209	+	-artifact-dir <path> Artifact staging directory (default: /tmp/pici-artifacts)
	210	+	-log-level <level> Log level: debug, info, warn, error`)
	211	+	}
	212	+
	213	+	func printMissingPicoHelp(cwd string) {
	214	+	fmt.Printf(`❌ Error: no pico.sh found in %s
	215	+
	216	+	To run local CI tasks, create a pico.sh script in your project root:
	217	+
	218	+	#!/usr/bin/env bash
	219	+	set -euo pipefail
	220	+
	221	+	# Run parallel steps using zmx
	222	+	zmx run lint -d <your lint command>
	223	+	zmx run test -d <your test command>
	224	+
	225	+	# Wait for all steps to finish
	226	+	zmx wait "*"
	227	+	printf "\x1b[32msuccess!\x1b[0m\n"
	228	+
	229	+	For full command documentation and daemon options, run:
	230	+	pici help
	231	+
	232	+	`, cwd)
	233	+	}
	234	+
163	235		func main() {
164	236		cfg, cmd, wantHelp := NewCfg()
165	237
166	238		cfg.Logger.Debug("setting up ci", "cfg", cfg)
167	239		cfg.Logger.Debug("running cmd", "cmd", cmd)
168	240
169		-	if wantHelp && (cmd == "runner" \|\| cmd == "monitor") {
170		-	if cmd == "runner" {
	241	+	if wantHelp {
	242	+	switch cmd {
	243	+	case "runner":
171	244		printRunnerHelp()
172		-	} else {
	245	+	case "monitor":
173	246		printMonitorHelp()
	247	+	default:
	248	+	printMainHelp()
174	249		}
175	250		return
176	251		}

...	...		@@ -204,9 +279,23 @@ func main() {
204	279		cfg.Logger.Debug("starting status updater")
205	280		case "orca":
206	281		cfg.Logger.Debug("starting orchestrator")
	282	+	case "help":
	283	+	printMainHelp()
	284	+	case "run", "":
	285	+	dest := ""
	286	+	if flag.NArg() > 0 {
	287	+	dest = flag.Arg(0)
	288	+	}
	289	+	if err := runLocal(cfg, dest); err != nil {
	290	+	cfg.Logger.Error("local run failed", "err", err)
	291	+	os.Exit(1)
	292	+	}
207	293		default:
208		-	cfg.Logger.Error("must provide command: runner, cancel, gc, monitor, status, or orca")
209		-	os.Exit(1)
	294	+	dest := cmd
	295	+	if err := runLocal(cfg, dest); err != nil {
	296	+	cfg.Logger.Error("local run failed", "err", err)
	297	+	os.Exit(1)
	298	+	}
210	299		}
211	300		}
212	301

...	...		@@ -362,7 +451,9 @@ func (w *WorkspaceRsync) Setup() error {
362	451		}
363	452
364	453		func (w *WorkspaceRsync) Cleanup() error {
365		-	// return os.RemoveAll(w.Dest)
	454	+	if w.Dest != "" {
	455	+	return os.RemoveAll(w.Dest)
	456	+	}
366	457		return nil
367	458		}
368	459

...	...		@@ -522,9 +613,13 @@ func (eng *JobEngine) Setup() error {
522	613		}
523	614
524	615		func (eng *JobEngine) Run(manifest string) error {
525		-	prefix := fmt.Sprintf("ci.%s.%s.", eng.Ev.Name, eng.JobID)
	616	+	domain := "ci"
	617	+	if eng.Ev != nil && eng.Ev.Type == "local" {
	618	+	domain = "local"
	619	+	}
	620	+	prefix := fmt.Sprintf("%s.%s.%s.", domain, eng.Ev.Name, eng.JobID)
526	621		// Child sessions use ".step." sub-prefix so zmx wait "*" inside pico.sh
527		-	// matches ci.<name>.<jobID>.step.* but NOT ci.<name>.<jobID>.runner.
	622	+	// matches <domain>.<name>.<jobID>.step.* but NOT <domain>.<name>.<jobID>.runner.
528	623		// This avoids a deadlock where the runner waits for itself.
529	624		childPrefix := prefix + "step."
530	625

...	...		@@ -45,6 +45,17 @@ func defaultWorkspaceFactory(cfg Cfg, logger slog.Logger, source string) Works
45	45		}
46	46		}
47	47
	48	+	type envList []string
	49	+
	50	+	func (e *envList) String() string {
	51	+	return strings.Join(*e, ", ")
	52	+	}
	53	+
	54	+	func (e *envList) Set(value string) error {
	55	+	e = append(e, value)
	56	+	return nil
	57	+	}
	58	+
48	59		type Cfg struct {
49	60		Logger *slog.Logger
50	61		Ctx context.Context

...	...		@@ -61,6 +72,7 @@ type Cfg struct {
61	72		IncludeRunning bool // emit running status updates in addition to terminal
62	73		HumanOutput bool // human-readable output instead of JSONL / slog
63	74		Wait bool // block until job completes, print history and summary
	75	+	EnvVars envList // custom environment variables passed via -e / -env
64	76		}
65	77
66	78		type Event struct {

...	...		@@ -80,6 +92,7 @@ func NewCfg() (*Cfg, string, bool) {
80	92		var monitorInterval time.Duration
81	93		var gcInterval time.Duration
82	94		var logLevel string
	95	+	var envVars envList
83	96		flag.StringVar(&keyLoc, "pk", "", "ssh private key used to authenticate with pico services")
84	97		flag.StringVar(&certLoc, "ck", "", "ssh certificate public key used to authenticate with pico services (only required if using ssh certificates)")
85	98		flag.StringVar(&artifactDir, "artifact-dir", "/tmp/pici-artifacts", "local directory to stage artifacts")

...	...		@@ -87,6 +100,8 @@ func NewCfg() (*Cfg, string, bool) {
87	100		flag.DurationVar(&monitorInterval, "monitor-interval", 5*time.Second, "interval for monitoring zmx sessions")
88	101		flag.DurationVar(&gcInterval, "gc-interval", 10*time.Minute, "interval for garbage collection in monitor (0 to disable)")
89	102		flag.StringVar(&logLevel, "log-level", "info", "log level: debug, info, warn, error")
	103	+	flag.Var(&envVars, "e", "environment variable in KEY=VAL format (can be specified multiple times)")
	104	+	flag.Var(&envVars, "env", "environment variable in KEY=VAL format (can be specified multiple times)")
90	105		var includeRunning bool
91	106		var human bool
92	107		var wait bool

...	...		@@ -551,9 +646,8 @@ func (eng *JobEngine) Run(manifest string) error {
551	646		if eng.Ev.Tag != "" {
552	647		cmdEnv = append(cmdEnv, fmt.Sprintf("PICI_TAG=%s", eng.Ev.Tag))
553	648		}
554		-
555		-	zmxPrefixStr := fmt.Sprintf("ZMX_SESSION_PREFIX=%s", childPrefix)
556		-	cmd := exec.Command("zmx", "run", runnerName, "-d", zmxPrefixStr, "bash", manifest)
	649	+	bashCmd := fmt.Sprintf("export ZMX_SESSION_PREFIX=%q; exec bash %q", childPrefix, manifest)
	650	+	cmd := exec.Command("zmx", "run", runnerName, "-d", "bash", "-c", bashCmd)
557	651		cmd.Env = cmdEnv
558	652		cmd.Dir = eng.Wk.GetDir()
559	653

...	...		@@ -610,14 +704,18 @@ func eventHandler(cfg Cfg, eventData Event) error {
610	704		Ev: eventData,
611	705		JobID: jobID,
612	706		}
	707	+	var runErr error
613	708		defer func() {
614		-	if err := eng.Cleanup(); err != nil {
615		-	cfg.Logger.Error("engine cleanup", "err", err)
	709	+	if runErr != nil \|\| cfg.Wait {
	710	+	if err := eng.Cleanup(); err != nil {
	711	+	cfg.Logger.Error("engine cleanup", "err", err)
	712	+	}
616	713		}
617	714		}()
618	715
619	716		fmt.Fprintf(os.Stdout, "📦 syncing workspace %s\n", eventData.Workspace) //nolint:errcheck
620	717		if err := eng.Setup(); err != nil {
	718	+	runErr = err
621	719		return fmt.Errorf("setup: %w", err)
622	720		}
623	721		fmt.Fprintf(os.Stdout, "✅ workspace ready %s\n", eng.Wk.GetDir()) //nolint:errcheck

...	...		@@ -654,6 +752,7 @@ func eventHandler(cfg Cfg, eventData Event) error {
654	752
655	753		manifest, err := eng.FindManifest()
656	754		if err != nil {
	755	+	runErr = err
657	756		fmt.Fprintf(os.Stdout, "❌ %s\n\n", err) //nolint:errcheck
658	757		//nolint:errcheck
659	758		fmt.Fprint(os.Stdout, `Create a pico.sh script in your workspace root:

...	...		@@ -686,13 +785,15 @@ See: https://github.com/picosh/pici
686	785
687	786		fmt.Fprint(os.Stdout, "🏃 launching sessions...\n") //nolint:errcheck
688	787		if err := eng.Run(manifest); err != nil {
	788	+	runErr = err
689	789		return fmt.Errorf("run: %w", err)
690	790		}
691	791
692	792		fmt.Fprintln(os.Stdout, "✅ job launched") //nolint:errcheck
693	793
694	794		if cfg.Wait {
695		-	if err := waitAndReport(cfg, log, eventData.Name, jobID); err != nil {
	795	+	if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
	796	+	runErr = err
696	797		return fmt.Errorf("wait: %w", err)
697	798		}
698	799		return nil

...	...		@@ -707,9 +808,17 @@ See: https://github.com/picosh/pici
707	808
708	809		// waitAndReport polls the job's sessions until all complete, prints live
709	810		// progress to stdout, then dumps session history and a final summary.
710		-	func waitAndReport(cfg Cfg, log slog.Logger, name, jobID string) error {
711		-	prefix := "ci." + name + "." + jobID + "."
712		-	ticker := time.NewTicker(cfg.MonitorInterval)
	811	+	func waitAndReport(cfg Cfg, log slog.Logger, name, jobID, eventType string) error {
	812	+	domain := "ci"
	813	+	if eventType == "local" {
	814	+	domain = "local"
	815	+	}
	816	+	prefix := domain + "." + name + "." + jobID + "."
	817	+	interval := cfg.MonitorInterval
	818	+	if interval <= 0 {
	819	+	interval = 5 * time.Second
	820	+	}
	821	+	ticker := time.NewTicker(interval)
713	822		defer ticker.Stop()
714	823
715	824		// Handle ^C gracefully

...	...		@@ -726,9 +835,14 @@ func waitAndReport(cfg Cfg, log slog.Logger, name, jobID string) error {
726	835		var sessionOrder []string // track insertion order for deterministic output
727	836		var liveLines []string // last set of status lines printed (for overwrite)
728	837
	838	+	var done <-chan struct{}
	839	+	if cfg.Ctx != nil {
	840	+	done = cfg.Ctx.Done()
	841	+	}
	842	+
729	843		for {
730	844		select {
731		-	case <-cfg.Ctx.Done():
	845	+	case <-done:
732	846		return cfg.Ctx.Err()
733	847		case <-sigCh:
734	848		fmt.Fprintln(os.Stdout, "\n⏹ cancelled") //nolint:errcheck

...	...		@@ -876,9 +990,11 @@ func cleanSessionShort(name, prefix, repoName, jobID string) string {
876	990		short := strings.TrimPrefix(name, prefix)
877	991		// Strip "step." prefix added by child sessions
878	992		short = strings.TrimPrefix(short, "step.")
879		-	// Strip nested full prefix (e.g. runner named ci.name.jobID.runner)
880		-	nested := "ci." + repoName + "." + jobID + "."
881		-	short = strings.TrimPrefix(short, nested)
	993	+	// Strip nested full prefix (e.g. runner named ci.name.jobID.runner or local.name.jobID.runner)
	994	+	nestedCI := "ci." + repoName + "." + jobID + "."
	995	+	short = strings.TrimPrefix(short, nestedCI)
	996	+	nestedLocal := "local." + repoName + "." + jobID + "."
	997	+	short = strings.TrimPrefix(short, nestedLocal)
882	998		return short
883	999		}
884	1000

...	...		@@ -917,10 +1033,11 @@ func runMonitor(cfg *Cfg) error {
917	1033		defer ticker.Stop()
918	1034
919	1035		// Optional GC ticker — runs garbage collection on a separate interval.
920		-	var gcTicker *time.Ticker
	1036	+	var gcChan <-chan time.Time
921	1037		if cfg.GCInterval > 0 {
922		-	gcTicker = time.NewTicker(cfg.GCInterval)
	1038	+	gcTicker := time.NewTicker(cfg.GCInterval)
923	1039		defer gcTicker.Stop()
	1040	+	gcChan = gcTicker.C
924	1041		}
925	1042
926	1043		// Track per-job display state across ticks (for human output)

...	...		@@ -938,7 +1055,7 @@ func runMonitor(cfg *Cfg) error {
938	1055		if err := monitorTick(cfg, log, output, jobStates); err != nil {
939	1056		log.Error("monitor tick", "err", err)
940	1057		}
941		-	case <-gcTicker.C:
	1058	+	case <-gcChan:
942	1059		log.Debug("running periodic garbage collection")
943	1060		if err := runGC(cfg); err != nil {
944	1061		log.Error("periodic gc", "err", err)

...	...		@@ -1560,21 +1677,21 @@ func syncJobArtifacts(cfg Cfg, repoName, jobID string, log slog.Logger) error
1560	1677		}
1561	1678		sshArgs = fmt.Sprintf("-F ~/.ssh/config -i %s%s", cfg.KeyLocation, certFile)
1562	1679		}
1563		-	// Append "/" so rsync copies into the destination directory,
1564		-	// not as a subdirectory named after the source.
	1680	+	// Source is jobDir (without trailing slash) so rsync creates
	1681	+	// the jobID subdirectory inside the destination folder.
1565	1682		dest := event.ArtifactDest
1566	1683		if !strings.HasSuffix(dest, "/") {
1567	1684		dest += "/"
1568	1685		}
1569	1686		var cmd *exec.Cmd
1570	1687		if sshArgs != "" {
1571		-	cmd = exec.Command("rsync", "-e", sshArgs, "-rv", jobDir+"/", dest)
	1688	+	cmd = exec.Command("rsync", "-e", sshArgs, "-rv", jobDir, dest)
1572	1689		} else {
1573		-	cmd = exec.Command("rsync", "-rv", jobDir+"/", dest)
	1690	+	cmd = exec.Command("rsync", "-rv", jobDir, dest)
1574	1691		}
1575	1692		rsyncCmd := fmt.Sprintf("rsync %s %s %s",
1576	1693		strings.TrimLeft(cmd.Args[1], "-"),
1577		-	jobDir+"/", dest)
	1694	+	jobDir, dest)
1578	1695		log.Info("rsync", "cmd", rsyncCmd)
1579	1696		return runCmd(cmd, log)
1580	1697		}

...	...		@@ -1730,7 +1847,7 @@ func runGC(cfg *Cfg) error {
1730	1847
1731	1848		var toKill []string
1732	1849		for _, s := range sessions {
1733		-	if !strings.HasPrefix(s.Name, "ci.") {
	1850	+	if !strings.HasPrefix(s.Name, "ci.") && !strings.HasPrefix(s.Name, "local.") {
1734	1851		continue
1735	1852		}
1736	1853

...	...		@@ -1951,3 +2068,248 @@ func fmtDuration(created, ended string) string {
1951	2068		}
1952	2069		return fmt.Sprintf("%.1fs", secs)
1953	2070		}
	2071	+
	2072	+	func runLocal(cfg *Cfg, dest string) error {
	2073	+	cwd, err := os.Getwd()
	2074	+	if err != nil {
	2075	+	return fmt.Errorf("get working directory: %w", err)
	2076	+	}
	2077	+
	2078	+	picoPath := filepath.Join(cwd, "pico.sh")
	2079	+	if _, err := os.Stat(picoPath); os.IsNotExist(err) {
	2080	+	printMissingPicoHelp(cwd)
	2081	+	return fmt.Errorf("no pico.sh found in %s", cwd)
	2082	+	}
	2083	+
	2084	+	repoName := filepath.Base(cwd)
	2085	+	jobID := fmt.Sprintf("local-%d", time.Now().Unix())
	2086	+
	2087	+	eventData := &Event{
	2088	+	Type: "local",
	2089	+	Name: repoName,
	2090	+	JobID: jobID,
	2091	+	Workspace: cwd,
	2092	+	ArtifactDest: dest,
	2093	+	}
	2094	+
	2095	+	// Process -e / --env flags
	2096	+	for _, envPair := range cfg.EnvVars {
	2097	+	parts := strings.SplitN(envPair, "=", 2)
	2098	+	val := ""
	2099	+	if len(parts) == 2 {
	2100	+	val = parts[1]
	2101	+	}
	2102	+	key := parts[0]
	2103	+	switch key {
	2104	+	case "PICI_REPO":
	2105	+	eventData.Name = val
	2106	+	case "PICI_JOB":
	2107	+	eventData.JobID = val
	2108	+	case "PICI_EVENT":
	2109	+	eventData.Type = val
	2110	+	case "PICI_BRANCH":
	2111	+	eventData.Branch = val
	2112	+	case "PICI_COMMIT":
	2113	+	eventData.Commit = val
	2114	+	case "PICI_TAG":
	2115	+	eventData.Tag = val
	2116	+	}
	2117	+	_ = os.Setenv(key, val)
	2118	+	}
	2119	+
	2120	+	// Auto-detect git commit SHA and branch if not explicitly provided via -e
	2121	+	if eventData.Commit == "" {
	2122	+	eventData.Commit = detectGitCommit(cwd)
	2123	+	}
	2124	+	if eventData.Branch == "" {
	2125	+	eventData.Branch = detectGitBranch(cwd)
	2126	+	}
	2127	+
	2128	+	// Always block and output human format for local runs
	2129	+	cfg.Wait = true
	2130	+	cfg.HumanOutput = true
	2131	+
	2132	+	logger := cfg.Logger
	2133	+	if logger == nil {
	2134	+	logger = newLogger("ci", "info")
	2135	+	}
	2136	+	log := logger.With("repo", eventData.Name, "job_id", jobID, "mode", "local")
	2137	+
	2138	+	// Set up workspace in temp directory
	2139	+	wk := &WorkspaceRsync{
	2140	+	Cfg: cfg,
	2141	+	Logger: log,
	2142	+	Source: cwd,
	2143	+	}
	2144	+
	2145	+	eng := &JobEngine{
	2146	+	Logger: log,
	2147	+	Cfg: cfg,
	2148	+	Wk: wk,
	2149	+	Ev: eventData,
	2150	+	JobID: jobID,
	2151	+	}
	2152	+
	2153	+	defer func() {
	2154	+	if err := eng.Cleanup(); err != nil {
	2155	+	log.Error("cleanup workspace", "err", err)
	2156	+	}
	2157	+	}()
	2158	+
	2159	+	fmt.Fprintf(os.Stdout, "🚀 starting local job local.%s.%s\n", eventData.Name, jobID) //nolint:errcheck
	2160	+	fmt.Fprintln(os.Stdout, "📦 syncing workspace to temp directory...") //nolint:errcheck
	2161	+	if err := eng.Setup(); err != nil {
	2162	+	return fmt.Errorf("workspace setup: %w", err)
	2163	+	}
	2164	+	log.Debug("workspace directory", "dir", eng.Wk.GetDir())
	2165	+
	2166	+	// Store event.json in artifact dir
	2167	+	eventDir := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID)
	2168	+	artifactsDir := filepath.Join(eventDir, "artifacts")
	2169	+	if err := os.MkdirAll(artifactsDir, 0755); err != nil {
	2170	+	log.Error("create artifacts dir", "err", err)
	2171	+	} else {
	2172	+	eventBytes, _ := json.Marshal(eventData)
	2173	+	_ = os.WriteFile(filepath.Join(artifactsDir, "event.json"), eventBytes, 0644)
	2174	+	}
	2175	+
	2176	+	// Write attestation.json
	2177	+	hostname, _ := os.Hostname()
	2178	+	attestation := map[string]interface{}{
	2179	+	"runner": map[string]string{
	2180	+	"hostname": hostname,
	2181	+	"os": runtimeOS(),
	2182	+	"arch": runtimeArch(),
	2183	+	},
	2184	+	"provenance": map[string]string{
	2185	+	"repo": eventData.Name,
	2186	+	"branch": eventData.Branch,
	2187	+	"commit": eventData.Commit,
	2188	+	},
	2189	+	"workspace_checksum": eng.Wk.Checksum(),
	2190	+	}
	2191	+	attestationBytes, _ := json.Marshal(attestation)
	2192	+	_ = os.WriteFile(filepath.Join(artifactsDir, "attestation.json"), attestationBytes, 0644)
	2193	+
	2194	+	manifest, err := eng.FindManifest()
	2195	+	if err != nil {
	2196	+	return err
	2197	+	}
	2198	+
	2199	+	fmt.Fprintln(os.Stdout, "🏃 launching sessions...") //nolint:errcheck
	2200	+	if err := eng.Run(manifest); err != nil {
	2201	+	return fmt.Errorf("run: %w", err)
	2202	+	}
	2203	+
	2204	+	// Wait for completion & print live progress
	2205	+	if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
	2206	+	return fmt.Errorf("wait: %w", err)
	2207	+	}
	2208	+
	2209	+	// Generate and stage full HTML/txt artifacts and index
	2210	+	if err := stageLocalArtifacts(cfg, log, eventData.Name, jobID); err != nil {
	2211	+	log.Error("stage local artifacts", "err", err)
	2212	+	}
	2213	+
	2214	+	indexFile := filepath.Join(cfg.ArtifactDir, eventData.Name, jobID, "index.html")
	2215	+	fmt.Fprintf(os.Stdout, "📊 local html report: file://%s\n", indexFile) //nolint:errcheck
	2216	+
	2217	+	// Sync to destination if specified
	2218	+	if dest != "" {
	2219	+	targetDest := strings.TrimSuffix(dest, "/") + "/" + jobID
	2220	+	fmt.Fprintf(os.Stdout, "🔄 rsyncing artifacts to %s...\n", dest) //nolint:errcheck
	2221	+	if err := syncJobArtifacts(cfg, eventData.Name, jobID, log); err != nil {
	2222	+	return fmt.Errorf("sync artifacts: %w", err)
	2223	+	}
	2224	+	fmt.Fprintf(os.Stdout, "✅ artifacts rsynced to %s\n", targetDest) //nolint:errcheck
	2225	+	}
	2226	+
	2227	+	return nil
	2228	+	}
	2229	+
	2230	+	func stageLocalArtifacts(cfg Cfg, log slog.Logger, repoName, jobID string) error {
	2231	+	listOutput, err := exec.Command("zmx", "list").CombinedOutput()
	2232	+	if err != nil {
	2233	+	return fmt.Errorf("zmx list: %w", err)
	2234	+	}
	2235	+	sessions := parseZMXList(string(listOutput))
	2236	+	var localSessions []SessionInfo
	2237	+	for _, s := range sessions {
	2238	+	if strings.HasPrefix(s.Name, "local.") {
	2239	+	localSessions = append(localSessions, s)
	2240	+	}
	2241	+	}
	2242	+	prefix := fmt.Sprintf("local.%s.%s.", repoName, jobID)
	2243	+
	2244	+	var jobSessions []SessionInfo
	2245	+	for _, s := range localSessions {
	2246	+	if strings.HasPrefix(s.Name, prefix) {
	2247	+	s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
	2248	+	jobSessions = append(jobSessions, s)
	2249	+	}
	2250	+	}
	2251	+
	2252	+	for _, s := range jobSessions {
	2253	+	sessionStatus := "running"
	2254	+	sessionDuration := fmtDuration(s.Created, fmt.Sprintf("%d", time.Now().Unix()))
	2255	+	sessionExitCode := ""
	2256	+	if s.Ended != "" {
	2257	+	sessionDuration = fmtDuration(s.Created, s.Ended)
	2258	+	if s.ExitCode == "0" {
	2259	+	sessionStatus = "success"
	2260	+	sessionExitCode = "0"
	2261	+	} else {
	2262	+	sessionStatus = "failed"
	2263	+	sessionExitCode = s.ExitCode
	2264	+	}
	2265	+	}
	2266	+
	2267	+	html, err := fetchHistoryHTML(s.Name, repoName, jobID, sessionStatus, sessionDuration, sessionExitCode)
	2268	+	if err == nil {
	2269	+	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, html, ".html")
	2270	+	}
	2271	+	plain, err := fetchHistoryPlain(s.Name)
	2272	+	if err == nil {
	2273	+	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, s.Short, plain, ".txt")
	2274	+	}
	2275	+	}
	2276	+
	2277	+	// Write published sentinel
	2278	+	exitCode, status := resolveJobExitCode(jobSessions)
	2279	+	sentinel := filepath.Join(cfg.ArtifactDir, repoName, jobID, "artifacts", "published.json")
	2280	+	published := map[string]interface{}{
	2281	+	"status": status,
	2282	+	"exit_code": exitCode,
	2283	+	"job_id": jobID,
	2284	+	"finished_at": time.Now().UTC().Format(time.RFC3339),
	2285	+	}
	2286	+	publishedJSON, _ := json.Marshal(published)
	2287	+	_ = os.WriteFile(sentinel, publishedJSON, 0644)
	2288	+
	2289	+	indexHTML, indexTXT := generateJobIndex(cfg.ArtifactDir, repoName, jobID, jobSessions)
	2290	+	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexHTML, ".html")
	2291	+	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "index", indexTXT, ".txt")
	2292	+	if styles, err := loadStyles(); err == nil {
	2293	+	_ = stageArtifact(cfg.ArtifactDir, repoName, jobID, "styles", styles, ".css")
	2294	+	}
	2295	+
	2296	+	return nil
	2297	+	}
	2298	+
	2299	+	func detectGitCommit(dir string) string {
	2300	+	cmd := exec.Command("git", "-C", dir, "rev-parse", "HEAD")
	2301	+	out, err := cmd.Output()
	2302	+	if err != nil {
	2303	+	return ""
	2304	+	}
	2305	+	return strings.TrimSpace(string(out))
	2306	+	}
	2307	+
	2308	+	func detectGitBranch(dir string) string {
	2309	+	cmd := exec.Command("git", "-C", dir, "rev-parse", "--abbrev-ref", "HEAD")
	2310	+	out, err := cmd.Output()
	2311	+	if err != nil \|\| strings.TrimSpace(string(out)) == "HEAD" {
	2312	+	return ""
	2313	+	}
	2314	+	return strings.TrimSpace(string(out))
	2315	+	}

...	...		@@ -25,17 +25,20 @@ func TestE2E_RunnerWithZMXSessions(t *testing.T) {
25	25		if testing.Short() {
26	26		t.Skip("skip integration test")
27	27		}
28		-	if _, err := exec.LookPath("zmx"); err != nil {
	28	+	zmxPath, err := exec.LookPath("zmx")
	29	+	if err != nil {
29	30		t.Skip("zmx not found, skipping integration test")
30	31		}
	32	+	zmxDir := filepath.Dir(zmxPath)
31	33
32	34		// 1. Create workspace with pico.sh that spawns zmx sessions
33	35		workspaceDir := t.TempDir()
34		-	picoSh := `#!/usr/bin/env bash
	36	+	picoSh := fmt.Sprintf(`#!/usr/bin/env bash
35	37		set -e
	38	+	export PATH="%s:$PATH"
36	39		zmx run step1 echo "hello from step1"
37	40		zmx run step2 echo "hello from step2"
38		-	`
	41	+	`, zmxDir)
39	42		if err := os.WriteFile(filepath.Join(workspaceDir, "pico.sh"), []byte(picoSh), 0755); err != nil {
40	43		t.Fatalf("write pico.sh: %v", err)
41	44		}

...	...		@@ -43,9 +46,11 @@ zmx run step2 echo "hello from step2"
43	46		// 2. Create config
44	47		artifactDir := t.TempDir()
45	48		ctx, cancel := context.WithCancel(context.Background())
	49	+	testJobID := fmt.Sprintf("j-%d", time.Now().UnixNano()%100000000)
46	50		event := Event{
47	51		Type: "build",
48	52		Name: "test-repo",
	53	+	JobID: testJobID,
49	54		Workspace: workspaceDir,
50	55		}
51	56		eventJSON, _ := json.Marshal(event)

...	...		@@ -109,7 +114,7 @@ zmx run step2 echo "hello from step2"
109	114		}
110	115
111	116		// Ignore statuses from unrelated jobs (e.g., leftover sessions from previous tests)
112		-	if p.Name != "test-repo" {
	117	+	if p.Name != "test-repo" \|\| p.JobID != testJobID {
113	118		continue
114	119		}
115	120

...	...		@@ -690,3 +695,56 @@ func TestResolveJobExitCode(t *testing.T) {
690	695		})
691	696		}
692	697		}
	698	+
	699	+	func TestRunLocal_MissingPico(t *testing.T) {
	700	+	tempDir := t.TempDir()
	701	+	origWd, _ := os.Getwd()
	702	+	defer func() { _ = os.Chdir(origWd) }()
	703	+	_ = os.Chdir(tempDir)
	704	+
	705	+	cfg := &Cfg{
	706	+	ArtifactDir: t.TempDir(),
	707	+	}
	708	+	err := runLocal(cfg, "")
	709	+	if err == nil {
	710	+	t.Fatal("expected error when pico.sh is missing")
	711	+	}
	712	+	}
	713	+
	714	+	func TestRunLocal_EnvOverrides(t *testing.T) {
	715	+	tempDir := t.TempDir()
	716	+	origWd, _ := os.Getwd()
	717	+	defer func() { _ = os.Chdir(origWd) }()
	718	+	_ = os.Chdir(tempDir)
	719	+
	720	+	picoContent := `#!/usr/bin/env bash
	721	+	echo "hello from pico"
	722	+	`
	723	+	if err := os.WriteFile("pico.sh", []byte(picoContent), 0755); err != nil {
	724	+	t.Fatal(err)
	725	+	}
	726	+
	727	+	cfg := &Cfg{
	728	+	ArtifactDir: t.TempDir(),
	729	+	EnvVars: envList{"PICI_REPO=custom-repo", "CUSTOM_VAR=hello"},
	730	+	}
	731	+
	732	+	_ = runLocal(cfg, "")
	733	+
	734	+	if os.Getenv("CUSTOM_VAR") != "hello" {
	735	+	t.Errorf("expected CUSTOM_VAR to be hello, got %q", os.Getenv("CUSTOM_VAR"))
	736	+	}
	737	+	}
	738	+
	739	+	func TestDetectGitCommit(t *testing.T) {
	740	+	cwd, err := os.Getwd()
	741	+	if err != nil {
	742	+	t.Fatal(err)
	743	+	}
	744	+	commit := detectGitCommit(cwd)
	745	+	if commit == "" {
	746	+	t.Log("git commit sha not detected (not a git repository or git unavailable)")
	747	+	} else {
	748	+	t.Logf("detected git commit sha: %s", commit)
	749	+	}
	750	+	}

pici

Patchset 132.1 on 2026-08-08T17:48:30Z · commit 0224564

Patchset `132.1` on 2026-08-08T17:48:30Z · commit 0224564