fix(hooks): report what a concurrent-group abort cut short

A command that fails to *start* aborts its whole group: the siblings
already running are killed and the ones after it never spawn. The
deferred report saw none of that — `record_failure` derives `skipped`
from the steps after the group, which can't see inside it — so a group
of `a`, `b`, `c` whose `b` fails to expand reported `user:b did not run`
with no clause at all, the same silence #3858 removed one level up.

`StepFailure` now carries the group's own casualties, in two lists
rather than one: a killed sibling ran partway and may have had side
effects, so reporting it as "skipped" would say it never ran. A sibling
that had already finished on its own (`try_wait`) is neither — it
completed, and now resolves its trace with its real status instead of
`false`.

    ▲ Background post-merge hook for main failed: user:b did not run; stopped a; skipped c

Also pins the parent commit's unnamed-step label fix with a test:
`{{ vars.never_set }}` is a cheap deterministic setup failure, so the
fallback that rendered `user:user post-merge hook` is reachable from the
suite after all.

src/commands/hook_failure.rs

@@ -12,8 +12,9 @@
 //! This module is the channel that closes that gap. The runner appends one JSON
 //! line per aborted pipeline to `.git/wt/hook-failures.jsonl`; the next
 //! foreground `wt` invocation in that repo drains the file and prints one
-//! warning per record, naming the step that failed and the steps its abort
-//! skipped, before the command's own output.
+//! warning per record, naming the step that failed, the steps its abort
+//! skipped, and any concurrent-group sibling it cut short, before the command's
+//! own output.
 //!
 //! # Contracts
 //!
@@ -76,6 +77,13 @@ pub struct HookFailure {
     /// Exit code of the failed step. `None` when the pipeline aborted before
     /// running it (template expansion, log creation, spawn).
     pub exit: Option<i32>,
+    /// Display names of the commands the abort cut short — the concurrent-group
+    /// siblings that were already running when this command failed to start.
+    /// Kept apart from `skipped` because these ran partway and may have had
+    /// side effects. Defaulted so a record written by an older `wt` still
+    /// reports.
+    #[serde(default)]
+    pub stopped: Vec<String>,
     /// Display names of the steps the abort skipped, in pipeline order.
     pub skipped: Vec<String>,
     /// Absolute path to the failed step's own output log, when it got far
@@ -94,19 +102,21 @@ impl HookFailure {
             Some(code) => format!("exited {code}"),
             None => "did not run".to_string(),
         };
-        let skipped = if self.skipped.is_empty() {
-            String::new()
-        } else {
-            let names = self
-                .skipped
+        let clause = |verb: &str, labels: &[String]| {
+            if labels.is_empty() {
+                return String::new();
+            }
+            let names = labels
                 .iter()
                 .map(|n| cformat!("<bold>{n}</>"))
                 .collect::<Vec<_>>()
                 .join(", ");
-            format!("; skipped {names}")
+            format!("; {verb} {names}")
         };
+        let stopped = clause("stopped", &self.stopped);
+        let skipped = clause("skipped", &self.skipped);
         cformat!(
-            "Background <bold>{hook_type}</> hook for <bold>{branch}</> failed: <bold>{source}:{failed}</> {outcome}{skipped}"
+            "Background <bold>{hook_type}</> hook for <bold>{branch}</> failed: <bold>{source}:{failed}</> {outcome}{stopped}{skipped}"
         )
     }
 }
@@ -222,6 +232,7 @@ mod tests {
             branch: "main".to_string(),
             failed: "sync".to_string(),
             exit: Some(1),
+            stopped: Vec::new(),
             skipped: vec!["push".to_string()],
             log: None,
         }
@@ -300,6 +311,24 @@ mod tests {
         );
     }
 
+    /// A command that fails to start takes its concurrent-group siblings with
+    /// it two ways: the ones already running are killed, the ones after it
+    /// never start. Reporting both as "skipped" would say a half-run command
+    /// never ran, so they get their own clause.
+    #[test]
+    fn headline_separates_the_commands_it_cut_short_from_the_ones_that_never_ran() {
+        let mut f = failure();
+        f.failed = "b".to_string();
+        f.exit = None;
+        f.stopped = vec!["a".to_string()];
+        f.skipped = vec!["c".to_string(), "push".to_string()];
+        let plain = anstream::adapter::strip_str(&f.headline()).to_string();
+        assert_eq!(
+            plain,
+            "Background post-merge hook for main failed: user:b did not run; stopped a; skipped c, push"
+        );
+    }
+
     #[test]
     fn take_pending_drains_records_and_leaves_none_behind() {
         let dir = tempfile::tempdir().unwrap();

src/commands/run_pipeline.rs

@@ -27,7 +27,9 @@
 //!
 //! **Concurrent groups** spawn each child as soon as its own template is
 //! expanded, then wait for every child before proceeding. If any child fails,
-//! the group is reported as failed, but all children are allowed to finish.
+//! the group is reported as failed, but all children are allowed to finish —
+//! unless a command fails *before* spawning (expansion, log creation, spawn),
+//! which kills the siblings already running and leaves the rest unspawned.
 //! Expansion runs in a single sequential loop in command order — each command
 //! is expanded immediately before its own child is spawned (expansion may read
 //! git config, so order matters for `vars.*`), so a later command's expansion
@@ -129,6 +131,13 @@ struct StepFailure {
     /// Log-file stem for that command (`name`, or `cmd-{index}`), used to point
     /// the report at its output. `None` when no log file was created.
     log_name: Option<String>,
+    /// Commands of the same concurrent group that were still running when this
+    /// one failed to start, and so were killed. Empty for a serial step, and
+    /// for a group whose commands all spawned (those are waited out).
+    stopped: Vec<String>,
+    /// Commands of the same concurrent group that never started because of the
+    /// abort. The steps *after* the group are added by [`record_failure`].
+    skipped: Vec<String>,
     error: anyhow::Error,
 }
 
@@ -137,6 +146,8 @@ impl StepFailure {
         Self {
             label: label.into(),
             log_name: log_name.map(str::to_owned),
+            stopped: Vec::new(),
+            skipped: Vec::new(),
             error,
         }
     }
@@ -203,10 +214,10 @@ fn record_failure(
     step_index: usize,
     failure: &StepFailure,
 ) {
-    let skipped: Vec<String> = spec.steps[step_index + 1..]
-        .iter()
-        .flat_map(step_labels)
-        .collect();
+    // A concurrent-group abort skips the rest of its own group before it skips
+    // the steps that follow, so the group's members lead the list.
+    let mut skipped = failure.skipped.clone();
+    skipped.extend(spec.steps[step_index + 1..].iter().flat_map(step_labels));
     let log = failure.log_name.as_ref().map(|name| {
         HookLog::hook(spec.source, spec.hook_type, name).path(&spec.log_dir, &spec.branch)
     });
@@ -218,6 +229,7 @@ fn record_failure(
             branch: spec.branch.clone(),
             failed: failure.label.clone(),
             exit: failure.error.exit_code(),
+            stopped: failure.stopped.clone(),
             skipped,
             log,
         },
@@ -346,12 +358,49 @@ fn wait_resolving(
     }
 }
 
+/// A running command of a concurrent group: its hook name, log stem, expanded
+/// template, child, and the trace guard the waiter resolves.
+type SpawnedCommand = (Option<String>, String, String, Child, CommandTrace);
+
+/// Expand one concurrent-group command and spawn its child.
+///
+/// Split out of [`run_concurrent_group`] so every setup failure — log creation,
+/// expansion, spawn — is one `?` there, leaving the loop free to attribute the
+/// error to the command's position in the group.
+fn spawn_group_command(
+    cmd: &super::pipeline_spec::PipelineCommandSpec,
+    spec: &PipelineSpec,
+    repo: &Repository,
+    cmd_index: &mut usize,
+) -> Result<SpawnedCommand, StepFailure> {
+    let log_name = command_log_name(cmd.name.as_deref(), *cmd_index);
+    // Before expansion the raw template is what the user wrote, so it stands in
+    // for a command with no name (as `step_labels` does for a skipped one).
+    let label = cmd.name.as_deref().unwrap_or(&cmd.template).to_string();
+    let log_file =
+        create_command_log(spec, &log_name).map_err(|e| StepFailure::new(e, &label, None))?;
+    let cmd_ctx = step_context(&spec.context, cmd.name.as_deref());
+    let expanded = expand_shell_template(&cmd.template, &cmd_ctx, repo, &cmd.template_name)
+        .map_err(|e| StepFailure::new(e, &label, Some(&log_name)))?;
+    let label = cmd.name.as_deref().unwrap_or(&expanded).to_string();
+    let cmd_json = cmd_ctx.to_json();
+    let (child, trace) = spawn_shell_command(&expanded, &spec.worktree_path, &cmd_json, log_file)
+        .map_err(|e| StepFailure::new(e, &label, Some(&log_name)))?;
+    *cmd_index += 1;
+    Ok((cmd.name.clone(), log_name, expanded, child, trace))
+}
+
 /// Spawn all commands in a concurrent group, then wait for all.
 ///
 /// Waits every spawned child before returning. If any failed, the first
 /// failure (in spawn order) is returned, matching the serial-step bail
 /// format. Per-command output already lives in each command's log file.
 ///
+/// A command that fails to *start* is the exception: the group can't run as
+/// specified, so the siblings already running are killed and the ones after it
+/// never spawn. The returned failure carries both sets, since the step index
+/// [`record_failure`] works from can't see inside the group.
+///
 /// When `WORKTRUNK_TEST_SERIAL_CONCURRENT=1` is set, each command's child is
 /// awaited before the next is spawned so output ordering is deterministic for
 /// snapshot tests. The serial path bails on the first failure rather than
@@ -364,53 +413,61 @@ fn run_concurrent_group(
     cmd_index: &mut usize,
 ) -> Result<(), StepFailure> {
     let serial = super::force_serial_concurrent();
-    let mut children: Vec<(Option<String>, String, String, Child, CommandTrace)> =
+    let mut children: Vec<SpawnedCommand> =
         Vec::with_capacity(if serial { 0 } else { commands.len() });
 
     // Spawn (and, in serial mode, run) each command. Wrapped so that a mid-loop
     // error — a setup `?` or a spawn failure for a later command — tears down
     // the children already spawned this group rather than dropping them with
-    // unresolved trace guards (and as unreaped orphans).
-    let spawn_result = (|| -> Result<(), StepFailure> {
-        for cmd in commands {
-            let log_name = command_log_name(cmd.name.as_deref(), *cmd_index);
-            let label = cmd.name.as_deref().unwrap_or(&cmd.template).to_string();
-            let log_file = create_command_log(spec, &log_name)
-                .map_err(|e| StepFailure::new(e, &label, None))?;
-            let cmd_ctx = step_context(&spec.context, cmd.name.as_deref());
-            let expanded = expand_shell_template(&cmd.template, &cmd_ctx, repo, &cmd.template_name)
-                .map_err(|e| StepFailure::new(e, &label, Some(&log_name)))?;
-            let label = cmd.name.as_deref().unwrap_or(&expanded).to_string();
-            let cmd_json = cmd_ctx.to_json();
-            let (mut child, mut trace) =
-                spawn_shell_command(&expanded, &spec.worktree_path, &cmd_json, log_file)
-                    .map_err(|e| StepFailure::new(e, &label, Some(&log_name)))?;
-            *cmd_index += 1;
+    // unresolved trace guards (and as unreaped orphans). The error carries the
+    // failing command's position in the group, which is what lets the teardown
+    // name the siblings it cut short and the ones that never started.
+    let spawn_result = (|| -> Result<(), (usize, StepFailure)> {
+        for (group_index, cmd) in commands.iter().enumerate() {
+            let (name, log_name, expanded, mut child, mut trace) =
+                spawn_group_command(cmd, spec, repo, cmd_index)
+                    .map_err(|failure| (group_index, failure))?;
 
             if serial {
+                let label = name.as_deref().unwrap_or(&expanded).to_string();
                 let status = wait_resolving(&mut child, &mut trace, &expanded)
-                    .map_err(|e| StepFailure::new(e, &label, Some(&log_name)))?;
+                    .map_err(|e| (group_index, StepFailure::new(e, &label, Some(&log_name))))?;
                 if !status.success() {
-                    return Err(StepFailure::new(
-                        failure_error(&status, &label),
-                        &label,
-                        Some(&log_name),
+                    return Err((
+                        group_index,
+                        StepFailure::new(failure_error(&status, &label), &label, Some(&log_name)),
                     ));
                 }
             } else {
-                children.push((cmd.name.clone(), log_name, expanded, child, trace));
+                children.push((name, log_name, expanded, child, trace));
             }
         }
         Ok(())
     })();
 
-    if let Err(e) = spawn_result {
-        for (_, _, _, mut child, mut trace) in children {
-            let _ = child.kill();
-            let _ = child.wait();
-            trace.complete(false);
+    if let Err((group_index, mut failure)) = spawn_result {
+        // Nothing else reports these: the killed siblings ran only partway and
+        // the later commands never spawned, while the group's own step index is
+        // all `record_failure` can see. Both go on the failure so the deferred
+        // notice names them.
+        for (name, _, expanded, mut child, mut trace) in children {
+            match child.try_wait() {
+                // Finished on its own before the abort reached it, so it ran to
+                // completion and isn't something the abort cut short.
+                Ok(Some(status)) => trace.complete(status.success()),
+                _ => {
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    trace.complete(false);
+                    failure.stopped.push(name.unwrap_or(expanded));
+                }
+            }
         }
-        return Err(e);
+        failure.skipped = commands[group_index + 1..]
+            .iter()
+            .map(|cmd| cmd.name.clone().unwrap_or_else(|| cmd.template.clone()))
+            .collect();
+        return Err(failure);
     }
 
     wait_first_error(children.into_iter().map(

tests/integration_tests/user_hooks.rs

@@ -703,6 +703,110 @@ push = "echo PUSHED > pushed.txt"
     );
 }
 
+/// A list-form hook has no name, so the report falls back to what the user
+/// wrote. The fallback has to be the template itself: the label the runner
+/// carries for expansion errors is already `"{source} {hook_type} hook"`, and
+/// the report prefixes the source again — reporting `user:user post-merge
+/// hook`.
+#[rstest]
+fn test_background_failure_names_an_unnamed_step_by_its_command(mut repo: TestRepo) {
+    let feature_wt =
+        repo.add_worktree_with_commit("feature", "feature.txt", "feature content", "Add feature");
+
+    repo.write_test_config(
+        r#"post-merge = ["echo {{ vars.never_set }}", "echo AFTER > after.txt"]
+"#,
+    );
+
+    let mut merge = make_snapshot_cmd(
+        &repo,
+        "merge",
+        &["main", "--yes", "--no-remove"],
+        Some(&feature_wt),
+    );
+    assert!(merge.status().unwrap().success());
+
+    wait_for_file_content(
+        &resolve_git_common_dir(repo.root_path())
+            .join("wt")
+            .join("hook-failures.jsonl"),
+    );
+
+    let output = make_snapshot_cmd(&repo, "list", &[], None)
+        .output()
+        .unwrap();
+    let stderr = anstream::adapter::strip_str(&String::from_utf8_lossy(&output.stderr)).to_string();
+    assert!(
+        stderr.contains("user:echo {{ vars.never_set }} did not run"),
+        "the unnamed step must be named by its own command, got:\n{stderr}"
+    );
+}
+
+/// A concurrent group whose command fails to *start* takes its siblings down
+/// two ways — the ones already running are killed, the ones after it never
+/// spawn — and neither is visible from the group's step index alone. The
+/// deferred report has to name both, or the abort is as silent inside a group
+/// as it was before #3858 across steps.
+///
+/// Unix-only for the `sleep`: killing the group leaves the shell's own child
+/// running on Windows, where it would hold the worktree open past the test.
+#[cfg(unix)]
+#[rstest]
+fn test_background_concurrent_group_failure_names_what_it_cut_short(mut repo: TestRepo) {
+    let feature_wt =
+        repo.add_worktree_with_commit("feature", "feature.txt", "feature content", "Add feature");
+
+    // `broken` fails at expansion — `vars.*` are read fresh per command, so a
+    // reference to one nothing set errors in the runner, after `slow` has
+    // already been spawned and before `later` is.
+    repo.write_test_config(
+        r#"post-merge = [
+    { slow = "sleep 30", broken = "echo {{ vars.never_set }}", later = "echo LATER > later.txt" },
+    "echo AFTER > after.txt"
+]
+"#,
+    );
+
+    let mut merge = make_snapshot_cmd(
+        &repo,
+        "merge",
+        &["main", "--yes", "--no-remove"],
+        Some(&feature_wt),
+    );
+    assert!(merge.status().unwrap().success());
+
+    let pending = resolve_git_common_dir(repo.root_path())
+        .join("wt")
+        .join("hook-failures.jsonl");
+    // The record landing at all means the group was torn down rather than left
+    // waiting on `slow`.
+    wait_for_file_content(&pending);
+
+    thread::sleep(SLEEP_FOR_ABSENCE_CHECK);
+    for never_ran in ["later.txt", "after.txt"] {
+        assert!(
+            !repo.root_path().join(never_ran).exists(),
+            "{never_ran} must not be written: neither the rest of the group nor the step after it runs"
+        );
+    }
+
+    let output = make_snapshot_cmd(&repo, "list", &[], None)
+        .output()
+        .unwrap();
+    // Stripped: each name is bolded, so the clauses only read whole in plain text.
+    let stderr = anstream::adapter::strip_str(&String::from_utf8_lossy(&output.stderr)).to_string();
+    for expected in [
+        "user:broken did not run",
+        "stopped slow",
+        "skipped later, echo AFTER > after.txt",
+    ] {
+        assert!(
+            stderr.contains(expected),
+            "expected {expected:?} in the deferred failure report, got:\n{stderr}"
+        );
+    }
+}
+
 // ============================================================================
 // User Pre-Remove Hook Tests
 // ============================================================================