Commit 8759922

fix: random monitoring issues

  1@@ -73,6 +73,7 @@ type Cfg struct {
  2 	HumanOutput         bool      // human-readable output instead of JSONL / slog
  3 	Wait                bool      // block until job completes, print history and summary
  4 	EnvVars             envList   // custom environment variables passed via -e / -env
  5+	SessionPrefix       string    // session prefix filter for monitor (default: "ci.")
  6 }
  7 
  8 type Event struct {
  9@@ -93,10 +94,12 @@ func NewCfg() (*Cfg, string, bool) {
 10 	var gcInterval time.Duration
 11 	var logLevel string
 12 	var envVars envList
 13+	var sessionPrefix string
 14 	flag.StringVar(&keyLoc, "pk", "", "ssh private key used to authenticate with pico services")
 15 	flag.StringVar(&certLoc, "ck", "", "ssh certificate public key used to authenticate with pico services (only required if using ssh certificates)")
 16 	flag.StringVar(&artifactDir, "artifact-dir", "/tmp/pici-artifacts", "local directory to stage artifacts")
 17 	flag.StringVar(&event, "event", "", "event JSON to run (alternative to reading from stdin)")
 18+	flag.StringVar(&sessionPrefix, "prefix", "ci.", "session prefix filter for monitor (default: ci.)")
 19 	flag.DurationVar(&monitorInterval, "monitor-interval", 5*time.Second, "interval for monitoring zmx sessions")
 20 	flag.DurationVar(&gcInterval, "gc-interval", 10*time.Minute, "interval for garbage collection in monitor (0 to disable)")
 21 	flag.StringVar(&logLevel, "log-level", "info", "log level: debug, info, warn, error")
 22@@ -134,6 +137,7 @@ func NewCfg() (*Cfg, string, bool) {
 23 		HumanOutput:         human,
 24 		Wait:                wait,
 25 		EnvVars:             envVars,
 26+		SessionPrefix:       sessionPrefix,
 27 	}, cmd, wantHelp
 28 }
 29 
 30@@ -614,8 +618,8 @@ func (eng *JobEngine) Setup() error {
 31 
 32 func (eng *JobEngine) Run(manifest string) error {
 33 	domain := "ci"
 34-	if eng.Ev != nil && eng.Ev.Type == "local" {
 35-		domain = "local"
 36+	if eng.Ev != nil && eng.Ev.Type != "" {
 37+		domain = eng.Ev.Type
 38 	}
 39 	prefix := fmt.Sprintf("%s.%s.%s.", domain, eng.Ev.Name, eng.JobID)
 40 	// Child sessions use ".step." sub-prefix so zmx wait "*" inside pico.sh
 41@@ -693,8 +697,8 @@ func eventHandler(cfg *Cfg, eventData *Event) error {
 42 
 43 	eventBytes, _ := json.Marshal(eventData)
 44 	domain := "ci"
 45-	if eventData.Type == "local" {
 46-		domain = "local"
 47+	if eventData.Type != "" {
 48+		domain = eventData.Type
 49 	}
 50 	fmt.Fprintf(os.Stdout, "🚀 starting job %s.%s.%s\n", domain, eventData.Name, jobID) //nolint:errcheck
 51 	fmt.Fprintf(os.Stdout, "   event: type=%s name=%s workspace=%s\n", eventData.Type, eventData.Name, eventData.Workspace) //nolint:errcheck
 52@@ -804,8 +808,8 @@ See: https://github.com/picosh/pici
 53 	}
 54 
 55 	domain = "ci"
 56-	if eventData.Type == "local" {
 57-		domain = "local"
 58+	if eventData.Type != "" {
 59+		domain = eventData.Type
 60 	}
 61 	session := fmt.Sprintf("%s.%s.%s.runner", domain, eventData.Name, jobID)
 62 	fmt.Fprintf(os.Stdout, "   zmx tail %s\n", session)    //nolint:errcheck
 63@@ -818,8 +822,8 @@ See: https://github.com/picosh/pici
 64 // progress to stdout, then dumps session history and a final summary.
 65 func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) error {
 66 	domain := "ci"
 67-	if eventType == "local" {
 68-		domain = "local"
 69+	if eventType != "" {
 70+		domain = eventType
 71 	}
 72 	prefix := domain + "." + name + "." + jobID + "."
 73 	interval := cfg.MonitorInterval
 74@@ -1180,7 +1184,7 @@ func monitorTick(cfg *Cfg, log *slog.Logger, output io.Writer, jobStates map[str
 75 		return fmt.Errorf("zmx list: %w", err)
 76 	}
 77 	sessions := parseZMXList(string(listOutput))
 78-	ciSessions := filterCISessions(sessions)
 79+	ciSessions := filterCISessions(cfg, sessions)
 80 
 81 	if len(ciSessions) == 0 {
 82 		log.Debug("no ci.* sessions found")
 83@@ -1420,11 +1424,18 @@ func fmtDurationTs(started, ended int64) string {
 84 	return fmt.Sprintf("%.1fs", secs)
 85 }
 86 
 87-// filterCISessions returns only sessions with ci. prefix.
 88-func filterCISessions(sessions []SessionInfo) []SessionInfo {
 89+// filterCISessions returns only sessions matching cfg.SessionPrefix (default "ci.").
 90+func filterCISessions(cfg *Cfg, sessions []SessionInfo) []SessionInfo {
 91+	prefix := "ci."
 92+	if cfg != nil && cfg.SessionPrefix != "" {
 93+		prefix = cfg.SessionPrefix
 94+	}
 95+	if !strings.HasSuffix(prefix, ".") {
 96+		prefix = prefix + "."
 97+	}
 98 	var filtered []SessionInfo
 99 	for _, s := range sessions {
100-		if strings.HasPrefix(s.Name, "ci.") {
101+		if strings.HasPrefix(s.Name, prefix) {
102 			filtered = append(filtered, s)
103 		}
104 	}

 1@@ -48,10 +48,10 @@ zmx run step2 echo "hello from step2"
 2 	ctx, cancel := context.WithCancel(context.Background())
 3 	testJobID := fmt.Sprintf("j-%d", time.Now().UnixNano()%100000000)
 4 	t.Cleanup(func() {
 5-		_ = exec.Command("zmx", "kill", "-f", fmt.Sprintf("ci.test-repo.%s", testJobID)).Run()
 6+		_ = exec.Command("zmx", "kill", "-f", fmt.Sprintf("test-ci.test-repo.%s", testJobID)).Run()
 7 	})
 8 	event := Event{
 9-		Type:      "ci",
10+		Type:      "test-ci",
11 		Name:      "test-repo",
12 		JobID:     testJobID,
13 		Workspace: workspaceDir,
14@@ -68,6 +68,7 @@ zmx run step2 echo "hello from step2"
15 		NewWorkspace:    defaultWorkspaceFactory,
16 		StatusOutput:    statusBuf,
17 		IncludeRunning:  true,
18+		SessionPrefix:   "test-ci.",
19 	}
20 
21 	// 3. Run the runner (fire-and-forget, exits quickly)
22@@ -184,8 +185,8 @@ zmx run step2 echo "hello from step2"
23 	if finalPayload == nil {
24 		t.Fatal("no final payload")
25 	}
26-	// Filter to only sessions with our job prefix to avoid picking up unrelated ci.* sessions
27-	expectedPrefix := fmt.Sprintf("ci.%s.%s.", finalPayload.Name, finalPayload.JobID)
28+	// Filter to only sessions with our job prefix to avoid picking up unrelated test-ci.* sessions
29+	expectedPrefix := fmt.Sprintf("test-ci.%s.%s.", finalPayload.Name, finalPayload.JobID)
30 	sessions := make([]SessionInfo, 0, len(finalPayload.Sessions))
31 	for _, s := range finalPayload.Sessions {
32 		if strings.HasPrefix(s.Name, expectedPrefix) {
33@@ -756,6 +757,13 @@ func TestDetectGitCommit(t *testing.T) {
34 }
35 
36 func TestWaitAndReport_Cancellation(t *testing.T) {
37+	if testing.Short() {
38+		t.Skip("skip integration test")
39+	}
40+	if _, err := exec.LookPath("zmx"); err != nil {
41+		t.Skip("zmx not found, skipping integration test")
42+	}
43+
44 	ctx, cancel := context.WithCancel(context.Background())
45 	cfg := &Cfg{
46 		Ctx:             ctx,
47@@ -764,9 +772,13 @@ func TestWaitAndReport_Cancellation(t *testing.T) {
48 	}
49 
50 	jobID := fmt.Sprintf("canceltest-%d", time.Now().UnixNano())
51-	prefix := "local.testrepo." + jobID + "."
52+	prefix := "test-local.testrepo." + jobID + "."
53 	runnerSession := prefix + "runner"
54 
55+	t.Cleanup(func() {
56+		_ = exec.Command("zmx", "kill", "-f", runnerSession).Run()
57+	})
58+
59 	cmd := exec.Command("zmx", "run", runnerSession, "-d", "sleep", "30")
60 	if err := cmd.Run(); err != nil {
61 		t.Fatalf("failed to start zmx session: %v", err)
62@@ -777,7 +789,7 @@ func TestWaitAndReport_Cancellation(t *testing.T) {
63 		cancel()
64 	}()
65 
66-	err := waitAndReport(cfg, nil, "testrepo", jobID, "local")
67+	err := waitAndReport(cfg, nil, "testrepo", jobID, "test-local")
68 	if err == nil {
69 		t.Error("expected error when waitAndReport is cancelled, got nil")
70 	}