feat(jobs): process-group kill for detached jobs; truncate list COMMAND column

docs/command-reference.md

@@ -108,7 +108,7 @@ agents can react without launching a diagnostic chain on every blip:
 - `--container`: override which container to exec into (default: session target container). Works in both modes.
 - `--script` executes the uploaded file directly when it has a shebang; otherwise it falls back to `sh <file> ...`.
 - `--shell` cannot be combined with `--script`.
-- `--detach`: launches the command in the background and returns immediately. Prints per-pod launch details including job id, pid, log path, and metadata path.
+- `--detach`: launches the command in the background and returns immediately. Prints per-pod launch details including job id, pid, log path, and metadata path. The command runs in its own process group (via `setsid`) so `okdev jobs stop` can terminate the entire process tree, not just the leader.
 - When `--log-dir` is used with multiple declared groups, logs are written into per-group subdirectories under the requested path. `--log-dir` has no effect with `--detach` (detached jobs log inside the pod).
 - Detached exec preserves argv to the remote process; it is not flattened back into a single shell string. Detached `--script` uploads a per-pod temp file, launches it, and removes that temp file after completion.
 - Detached jobs store combined stdout/stderr and JSON metadata under `/var/okdev/exec/` on the shared okdev-runtime volume, so logs and metadata survive a target-container crash (e.g. OOMKill) and remain readable through the sidecar. If that directory cannot be created or written by the container user (e.g. a user-supplied root-owned volume with a non-root container), the launcher falls back to the legacy `/tmp/okdev-exec/` with a notice on stderr — same lifetime semantics as before. For large outputs (training logs, profiles, dumps), prefer `--log-dir` on the caller or redirect inside your command to a session volume; the runtime volume is a node-local emptyDir and heavy writers can fill it.
@@ -133,7 +133,8 @@ agents can react without launching a diagnostic chain on every blip:
 - If the target container is gone (e.g. OOMKilled), listing and `jobs logs` automatically retry through the `okdev-sidecar` container, which mounts the same runtime volume.
 - When a command fails because the container is gone, the error carries the termination cause when it can be determined, e.g. `container "pytorch" terminated: OOMKilled (exit 137, finished 2026-07-05T06:32:11Z)`. The same hint is appended to `okdev exec` FAILED lines and single-pod exec errors.
 - Text output is grouped by logical job id, with one row per detached launch showing job id, summarized state, pod count, earliest start time, and original command.
-- JSON output includes both the logical job summary and the per-pod `podStates` records.
+- On a terminal the COMMAND column is truncated (with `…`) to the remaining terminal width so long training commands don't wrap rows; piped/redirected output and `--json` always carry the full command.
+- JSON output includes both the logical job summary and the per-pod `podStates` records (with `pgid` and `groupLive` — the count of live processes still in the job's process group — when available).
 - State values: `running` (wrapper alive and user command in flight), `exited` (user command finished; exit code is recorded in `podStates` / JSON), and `orphaned` (metadata still says `running` but the pid has exited or been recycled - typically the wrapper was `SIGKILL`ed or the container was restarted before the completion metadata could be written). Grouped text summaries render forms such as `running(1/2)`, `exited(2/2)`, and `failed(1/2)`.
 - When any pod fails to list its jobs, the command still prints the jobs it was able to collect from the healthy pods and reports the failures in a `FAILED:` footer (or an `errors` array in `--json` output); exit status is non-zero so scripts can detect partial failures.
 - With `spec.exec.fanoutMode: auto|gateway` and interpod SSH enabled, the per-pod listing routes through the in-cluster fanout gateway (one apiserver exec instead of one per pod). The listing is read-only, so any gateway problem silently falls back to direct per-pod queries.
@@ -167,8 +168,10 @@ agents can react without launching a diagnostic chain on every blip:
 ### `okdev jobs stop <job-id> [session]`
 
 - Stops every still-running pod in the logical job by sending `SIGTERM`, waiting 10 seconds, then sending `SIGKILL` to survivors.
-- Prints which pods were signaled during the stop flow.
-- Returns non-zero when any pod could not be queried or signaled, or if pods are still running after the stop attempt.
+- Signals the job's whole **process group** when available: detached jobs are launched via `setsid`, so children the command forked (e.g. `torchrun` workers) are signaled too — including stragglers whose leader already exited. Stop only reports success once no live group member remains, so a clean return means the process tree (and its GPU memory) is gone.
+- Group membership is verified via the job's `OKDEV_JOB_ID` environment marker before signaling, so recycled pids/pgids are never signaled by mistake. Jobs launched by older okdev versions (or in containers without `setsid`) fall back to leader-only signaling.
+- Prints which pods were signaled (`pgid=N` for group signals, `pid=N` for the fallback) during the stop flow.
+- Returns non-zero when any pod could not be queried or signaled, or if processes are still alive after the stop attempt.
 
 ### `okdev exec-jobs [session]`
 

internal/cli/exec_jobs.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -22,6 +23,8 @@ type execJobView struct {
 	Container string `json:"container"`
 	JobID     string `json:"jobId"`
 	PID       int    `json:"pid"`
+	PGID      int    `json:"pgid,omitempty"`
+	GroupLive int    `json:"groupLive,omitempty"`
 	State     string `json:"state"`
 	ExitCode  *int   `json:"exitCode,omitempty"`
 	LogPath   string `json:"logPath"`
@@ -74,14 +77,15 @@ func runExecJobs(ctx context.Context, client connect.ExecClient, namespace strin
 			fmt.Fprintln(out, "No detached exec jobs found")
 		}
 		if len(jobs) > 0 {
+			commandLimit := jobsListCommandLimit(out, jobs)
 			table := make([][]string, 0, len(jobs))
 			for _, job := range jobs {
 				table = append(table, []string{
 					job.JobID,
 					job.SummaryState,
 					fmt.Sprintf("%d", job.Pods),
 					job.StartedAt,
-					job.Command,
+					truncateTableCell(job.Command, commandLimit),
 				})
 			}
 			output.PrintTable(out, []string{"JOB ID", "STATE", "PODS", "STARTED", "COMMAND"}, table)
@@ -223,6 +227,8 @@ func appendDetachJobRows(rows []execJobView, jobs []detachMetadata, jobID, conta
 			Container: containerLabel,
 			JobID:     job.JobID,
 			PID:       job.PID,
+			PGID:      job.PGID,
+			GroupLive: job.GroupLive,
 			State:     job.State,
 			ExitCode:  job.ExitCode,
 			LogPath:   job.StdoutPath,
@@ -322,30 +328,97 @@ func earliestStartedAt(rows []execJobView) string {
 	return earliest
 }
 
+// jobsListCommandLimit bounds the COMMAND column so long training commands
+// don't wrap table rows on a terminal. Returns 0 (no limit) for non-TTY
+// output — piped consumers get the full command; --json is the structured
+// path either way.
+func jobsListCommandLimit(out io.Writer, jobs []logicalExecJobView) int {
+	if !isTerminalFD(out) {
+		return 0
+	}
+	return jobsListCommandLimitForWidth(terminalWidth(), jobs)
+}
+
+func jobsListCommandLimitForWidth(termWidth int, jobs []logicalExecJobView) int {
+	if termWidth <= 0 {
+		return 0
+	}
+	widths := []int{len("JOB ID"), len("STATE"), len("PODS"), len("STARTED")}
+	for _, job := range jobs {
+		for i, s := range []string{job.JobID, job.SummaryState, fmt.Sprintf("%d", job.Pods), job.StartedAt} {
+			if len(s) > widths[i] {
+				widths[i] = len(s)
+			}
+		}
+	}
+	fixed := 0
+	for _, w := range widths {
+		fixed += w
+	}
+	// PrintTable joins columns with two spaces; four separators precede the
+	// COMMAND column.
+	limit := termWidth - fixed - 4*2
+	// Keep enough of the command to recognize the job even on narrow
+	// terminals; a slightly wrapped row beats an unreadable stub.
+	if limit < 16 {
+		limit = 16
+	}
+	return limit
+}
+
+func truncateTableCell(s string, limit int) string {
+	if limit <= 0 {
+		return s
+	}
+	r := []rune(s)
+	if len(r) <= limit {
+		return s
+	}
+	return string(r[:limit-1]) + "…"
+}
+
 func detachJobsCommand() []string {
-	// Each output line is "<alive>\t<json>" where <alive> is 1 if the job's
-	// pid is still running and its /proc/<pid>/environ contains the
+	// Each output line is "<alive>\t<groupLive>\t<json>". <alive> is 1 if the
+	// job's pid is still running and its /proc/<pid>/environ contains the
 	// OKDEV_JOB_ID we set in the wrapper. Matching on an environment
 	// variable (rather than cmdline) is robust against PID reuse even when
-	// the job's cmdline is the arbitrary user command. The find(1) call
-	// also reaps stale .tmp files left behind by killed launchers/wrappers.
-	// Both the current metadata dir and the legacy /tmp location are scanned
-	// so jobs launched by an older okdev remain visible after an upgrade.
+	// the job's cmdline is the arbitrary user command. <groupLive> counts
+	// live (non-zombie) processes still in the job's process group — the
+	// leader may be gone while its children (e.g. torchrun workers) linger,
+	// which is exactly what `jobs stop` must clean up. The group is
+	// identity-checked via OKDEV_JOB_ID in a member's environ (inherited by
+	// all descendants) before counting: a PGID number can only be recycled
+	// once the whole group is dead, so a mismatch means an unrelated group.
+	// The find(1) call also reaps stale .tmp files left behind by killed
+	// launchers/wrappers. Both the current metadata dir and the legacy /tmp
+	// location are scanned so jobs launched by an older okdev remain
+	// visible after an upgrade.
 	script := `for dir in '` + detachMetadataDir + `' '` + legacyDetachMetadataDir + `'; do
   [ -d "$dir" ] || continue
   find "$dir" -maxdepth 1 -name '*.tmp' -type f -mmin +1 -delete 2>/dev/null || true
   for f in "$dir"/*.json; do
     [ -e "$f" ] || continue
     contents=$(cat "$f")
     pid=$(printf '%s' "$contents" | sed -n 's/.*"pid":\([0-9][0-9]*\).*/\1/p' | head -n1)
+    pgid=$(printf '%s' "$contents" | sed -n 's/.*"pgid":\([0-9][0-9]*\).*/\1/p' | head -n1)
     job_id=$(basename "$f" .json)
     alive=0
     if [ -n "$pid" ] && [ -r "/proc/$pid/environ" ]; then
       if tr '\0' '\n' < "/proc/$pid/environ" 2>/dev/null | grep -Fqx "OKDEV_JOB_ID=$job_id"; then
         alive=1
       fi
     fi
-    printf '%s\t%s\n' "$alive" "$contents"
+    glive=0
+    if [ -n "$pgid" ] && [ "$pgid" -gt 0 ] 2>/dev/null; then
+      members=$(awk -v g="$pgid" '{ pid=$1; line=$0; sub(/^[0-9]+ \(.*\) /, "", line); split(line, f, " "); if (f[1] != "Z" && f[3] == g) print pid }' /proc/[0-9]*/stat 2>/dev/null)
+      for m in $members; do
+        if tr '\0' '\n' < "/proc/$m/environ" 2>/dev/null | grep -Fqx "OKDEV_JOB_ID=$job_id"; then
+          glive=$(set -- $members; echo $#)
+          break
+        fi
+      done
+    fi
+    printf '%s\t%s\t%s\n' "$alive" "$glive" "$contents"
   done
 done
 `
@@ -407,12 +480,20 @@ func parseDetachMetadataLines(raw string) ([]detachMetadata, error) {
 		if strings.TrimSpace(line) == "" {
 			continue
 		}
-		// Lines from detachJobsCommand are "<alive>\t<json>"; tolerate
-		// raw JSON for older containers that ran a previous cli version.
+		// Lines from detachJobsCommand are "<alive>\t<groupLive>\t<json>";
+		// tolerate the older "<alive>\t<json>" shape and raw JSON so mixed
+		// cli versions against the same pod keep working.
 		alive := true
+		groupLive := 0
 		if tab := strings.IndexByte(line, '\t'); tab >= 0 {
 			alive = line[:tab] != "0"
 			line = line[tab+1:]
+			if tab2 := strings.IndexByte(line, '\t'); tab2 >= 0 {
+				if n, err := strconv.Atoi(strings.TrimSpace(line[:tab2])); err == nil && n >= 0 {
+					groupLive = n
+					line = line[tab2+1:]
+				}
+			}
 		}
 		line = strings.TrimSpace(line)
 		if line == "" {
@@ -422,6 +503,7 @@ func parseDetachMetadataLines(raw string) ([]detachMetadata, error) {
 		if err := json.Unmarshal([]byte(line), &job); err != nil {
 			return nil, fmt.Errorf("parse detach job metadata: %w", err)
 		}
+		job.GroupLive = groupLive
 		if job.JobID == "" || job.Pod == "" || job.PID <= 0 || job.MetaPath == "" {
 			return nil, errors.New("missing detach job metadata")
 		}

internal/cli/exec_jobs_test.go

@@ -221,8 +221,12 @@ func TestDetachJobsCommandReconcilesAndReapsTmpFiles(t *testing.T) {
 		"alive=1",
 		"alive=0",
 		"grep -Fqx \"OKDEV_JOB_ID=$job_id\"",
-		// Per-line output is "<alive>\t<json>".
-		"printf '%s\\t%s\\n' \"$alive\" \"$contents\"",
+		// Per-line output is "<alive>\t<groupLive>\t<json>".
+		"printf '%s\\t%s\\t%s\\n' \"$alive\" \"$glive\" \"$contents\"",
+		// Group liveness: count non-zombie members of the job's process
+		// group, identity-checked via the inherited OKDEV_JOB_ID env var.
+		`"pgid":\([0-9][0-9]*\)`,
+		"/proc/[0-9]*/stat",
 		// Best-effort cleanup of stale temp files left by killed wrappers.
 		"-name '*.tmp'",
 		"-mmin +1",
@@ -388,6 +392,59 @@ func TestIsContainerUnavailableExecError(t *testing.T) {
 	}
 }
 
+func TestParseDetachMetadataLinesParsesGroupLive(t *testing.T) {
+	raw := "1\t3\t{\"jobId\":\"job-a\",\"pod\":\"p\",\"container\":\"dev\",\"pid\":100,\"pgid\":100,\"metaPath\":\"/var/okdev/exec/job-a.json\",\"state\":\"running\"}\n" +
+		// Older two-field lines and raw JSON must keep parsing (mixed cli
+		// versions against the same pod).
+		"0\t{\"jobId\":\"job-b\",\"pod\":\"p\",\"container\":\"dev\",\"pid\":101,\"metaPath\":\"/tmp/okdev-exec/job-b.json\",\"state\":\"running\"}\n"
+	jobs, err := parseDetachMetadataLines(raw)
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if len(jobs) != 2 {
+		t.Fatalf("expected 2 jobs, got %+v", jobs)
+	}
+	if jobs[0].GroupLive != 3 || jobs[0].PGID != 100 {
+		t.Fatalf("expected groupLive=3 pgid=100, got %+v", jobs[0])
+	}
+	if jobs[1].GroupLive != 0 || jobs[1].State != "orphaned" {
+		t.Fatalf("expected legacy line orphaned with groupLive=0, got %+v", jobs[1])
+	}
+}
+
+func TestJobsListCommandLimitForWidth(t *testing.T) {
+	jobs := []logicalExecJobView{{
+		JobID:        "20260705T063211Z-deadbeef", // 25 chars
+		SummaryState: "running(4/4)",              // 12 chars
+		Pods:         4,
+		StartedAt:    "2026-07-05T06:32:11Z", // 20 chars
+	}}
+	// fixed = 25 + 12 + len("PODS")=4 + 20 + 4*2 = 69
+	if got := jobsListCommandLimitForWidth(120, jobs); got != 51 {
+		t.Fatalf("expected limit 51, got %d", got)
+	}
+	// Narrow terminals keep a recognizable floor instead of a useless stub.
+	if got := jobsListCommandLimitForWidth(60, jobs); got != 16 {
+		t.Fatalf("expected floor 16, got %d", got)
+	}
+	// Unknown width disables truncation.
+	if got := jobsListCommandLimitForWidth(0, jobs); got != 0 {
+		t.Fatalf("expected no limit, got %d", got)
+	}
+}
+
+func TestTruncateTableCell(t *testing.T) {
+	if got := truncateTableCell("short", 10); got != "short" {
+		t.Fatalf("unexpected: %q", got)
+	}
+	if got := truncateTableCell("torchrun --nnodes 4 --nproc-per-node 8 pretrain.py", 20); got != "torchrun --nnodes 4…" {
+		t.Fatalf("unexpected truncation: %q", got)
+	}
+	if got := truncateTableCell("anything", 0); got != "anything" {
+		t.Fatalf("limit 0 must disable truncation: %q", got)
+	}
+}
+
 func TestCollectDetachJobsFiltersByJobID(t *testing.T) {
 	client := &fakePdshExecClient{
 		outputs: map[string]string{