fix(engine): hand finished background shells to the model at step boundaries

codewhale-app#3233 item 1. In `codewhale exec`, a shell command that
auto-backgrounded never got its output to the model: completions were
drained only when the model stopped calling tools, appended without
another model round, and exec ends at the first TurnComplete. The DSH
parity runs saw 0 of 198 backgrounded outputs.

- One helper, Engine::deliver_shell_completions, replaces the three copies
  of drain + runtime message + status (turn start in send_message, and the
  final-answer and late drains in continuation).
- tool_batch calls it after a tool batch's results are appended, before
  the next model request. A job that finished while the model was working
  reaches it on its next step with no extra model round; the event is an
  appended user-role runtime message (no KV prefix change) recorded through
  add_session_message.
- No retry at final-answer time: 03ece7d8b1 removed that deliberately
  (timing-dependent token spend and flaky tests).
- Child engines no longer drain: they share the parent's shell manager and
  root session id, so their drains claimed the parent's unowned completions
  into the child transcript. ACP turns stay excluded.
- drain_finished_jobs_with_evidence_inner no longer claims child-owned jobs.
  It used to claim them, cut their retained output to a tail and then have
  the engine filter them out, so a child's later task_shell_wait lost
  output. The idle-wake predicate already ignored child-owned jobs.
- The background notices say when output arrives: before the model's next
  step, and a job still running when the turn ends is stopped unseen in a
  headless run. Child-owned jobs say their completion stays in task/status.
- deliver_shell_completions carries the known-limitation note.

Verified:
- Real exec run against a scripted local mock model (no provider traffic),
  same script on a baseline and a fixed build of this tree: bash
  `sleep 4; echo BGDONE-7f3` with timeout 1 (auto-backgrounds), then a 6 s
  foreground bash, then a final answer. Both builds make 3 model requests.
  Baseline request 3 never contains BGDONE; fixed request 3 carries it as a
  runtime_event message after the second tool result. Artifacts in
  CW/artifacts/exec-bg-completion-20261018/.
- scripts/dev-test.sh tui shell completion background: 730 run, 730 passed
  (including background_completion_after_a_turn_is_delivered_once_on_the_
  next_turn, child_owned_background_completion_is_not_delivered_to_parent,
  child_owned_background_completion_does_not_wake_parent,
  drain_finished_jobs_reports_once, background_start_advertises_task_
  status_completion, yolo_mode_does_not_prompt_for_background_shell).
- CI-form clippy -p codewhale-tui --all-targets --all-features -D warnings
  clean; cargo fmt --check clean; blocking-call budget 539 (unchanged);
  dead-code budget at budget; sync-changelog up to date.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01NBKbhx19BRbZtvWK8rVKjz

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- A background shell job that finishes while the agent is still working now
+  reaches the model before its next step, instead of only when the turn ends.
+  In `codewhale exec`, which stops at the end of the turn, that output used to
+  be lost. The background notice now says that a job still running when the
+  turn ends is stopped unseen in a headless run, so the model knows to wait
+  explicitly when its answer depends on it. Sub-agents no longer pick up the
+  parent's finished jobs, and the parent no longer consumes a sub-agent's.
 - `codewhale review --base <ref>` gives the reviewer the `base..HEAD` commit
   messages as the author's stated intent, ahead of the diff, so it stops
   flagging changes their commit messages already explain. The messages are

crates/tui/CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- A background shell job that finishes while the agent is still working now
+  reaches the model before its next step, instead of only when the turn ends.
+  In `codewhale exec`, which stops at the end of the turn, that output used to
+  be lost. The background notice now says that a job still running when the
+  turn ends is stopped unseen in a headless run, so the model knows to wait
+  explicitly when its answer depends on it. Sub-agents no longer pick up the
+  parent's finished jobs, and the parent no longer consumes a sub-agent's.
 - `codewhale review --base <ref>` gives the reviewer the `base..HEAD` commit
   messages as the author's stated intent, ahead of the diff, so it stops
   flagging changes their commit messages already explain. The messages are

crates/tui/src/core/engine.rs

@@ -6396,22 +6396,7 @@ impl Engine {
         // Deliver completions that arrived after the previous turn before the
         // next user request is sent. This keeps background shell work
         // model-visible without requiring an explicit wait/poll tool call.
-        let shell_completions = if self.is_acp_turn() {
-            Vec::new()
-        } else {
-            self.drain_shell_completion_events()
-        };
-        if !shell_completions.is_empty() {
-            self.add_session_message(crate::runtime_handoff::shell_completion_runtime_message(
-                &shell_completions,
-            ))
-            .await;
-            if let Some(status) =
-                crate::core::engine::turn_loop::shell_completion_status_text(&shell_completions, "")
-            {
-                let _ = self.send_event(Event::status(status)).await;
-            }
-        }
+        self.deliver_shell_completions("").await;
 
         // A person writing to a goal that only the runtime stopped (a failed
         // or timed-out continuation) is continuing the work: resume it as a

crates/tui/src/core/engine/tests/test_cases_19.rs

@@ -387,10 +387,10 @@ fn turn_metadata_carries_the_narrowing_only_on_a_narrowed_turn() {
 /// first, and it is delivered exactly once.
 ///
 /// This exercises the engine's own shell manager through the same
-/// `drain_shell_completion_events` both delivery sites use — the next-turn
-/// boundary drain in `Engine::send_message` and the late drain in the turn
-/// loop — so the exactly-once guarantee holds across them rather than within
-/// one of them.
+/// `drain_shell_completion_events` every delivery site uses — turn start in
+/// `Engine::send_message`, the step boundary after a tool batch, and the
+/// finishing drains in the turn loop — so the exactly-once guarantee holds
+/// across them rather than within one of them.
 #[tokio::test]
 async fn background_completion_after_a_turn_is_delivered_once_on_the_next_turn() {
     let tmp = tempdir().expect("tempdir");

crates/tui/src/core/engine/turn_loop.rs

@@ -1152,9 +1152,6 @@ impl Engine {
             .unwrap_or_default();
         completions
             .into_iter()
-            // Child-owned output stays in task/status for explicit child
-            // waits. Only unowned jobs belong in the parent model stream.
-            .filter(|completion| completion.event.owner_agent_id.is_none())
             .map(|mut completion| {
                 let tool_call_id =
                     format!("background-shell-completion-{}", completion.event.task_id);
@@ -1177,6 +1174,31 @@ impl Engine {
             .collect()
     }
 
+    /// Hand finished background shell jobs to the model as one runtime
+    /// message. Called at turn start, after each tool batch and when the turn
+    /// is finishing; returns whether anything was delivered.
+    ///
+    /// Known limitation: a job still running when the model ends its turn
+    /// reaches it only through the idle wake, on hosts that keep the Engine
+    /// alive; headless exec shuts down and stops it unseen. ACP turns and
+    /// child engines never receive runtime shell completions and use
+    /// `task_shell_wait` instead.
+    pub(super) async fn deliver_shell_completions(&mut self, timing: &str) -> bool {
+        if self.child_host.is_some() {
+            return false;
+        }
+        let completions = self.drain_shell_completion_events();
+        if completions.is_empty() {
+            return false;
+        }
+        self.add_session_message(shell_completion_runtime_message(&completions))
+            .await;
+        if let Some(status) = shell_completion_status_text(&completions, timing) {
+            let _ = self.send_event(Event::status(status)).await;
+        }
+        true
+    }
+
     /// Keep workers alive while their tracked background shell work is still
     /// running. This is deliberately owner-based and read-only: an unowned
     /// shell job cannot extend any worker heartbeat.

crates/tui/src/core/engine/turn_loop/continuation.rs

@@ -113,18 +113,7 @@ impl Engine {
                 return PhaseResult::Retry;
             }
 
-            let shell_completions = if self.is_acp_turn() {
-                Vec::new()
-            } else {
-                self.drain_shell_completion_events()
-            };
-            if !shell_completions.is_empty() {
-                self.add_session_message(shell_completion_runtime_message(&shell_completions))
-                    .await;
-                if let Some(status) = shell_completion_status_text(&shell_completions, "") {
-                    let _ = self.send_event(Event::status(status)).await;
-                }
-            }
+            self.deliver_shell_completions("").await;
 
             // Sub-agent completion handoff (issue #756). Resuming when
             // queued completions exist is correct; #3216 says do not wait
@@ -175,19 +164,7 @@ impl Engine {
             // arrived between the last hold check and now. If a child finished
             // while we were running the thinking-only check, surface its
             // sentinel rather than delaying it to the next turn.
-            let late_shell_completions = if self.is_acp_turn() {
-                Vec::new()
-            } else {
-                self.drain_shell_completion_events()
-            };
-            if !late_shell_completions.is_empty() {
-                self.add_session_message(shell_completion_runtime_message(&late_shell_completions))
-                    .await;
-                if let Some(status) = shell_completion_status_text(&late_shell_completions, "late")
-                {
-                    let _ = self.send_event(Event::status(status)).await;
-                }
-            }
+            self.deliver_shell_completions("late").await;
 
             if !self.is_acp_turn() && self.drain_subagent_completion_events("late").await > 0 {
                 let _ = self

crates/tui/src/core/engine/turn_loop/tool_batch.rs

@@ -234,6 +234,8 @@ impl Engine {
                 .await;
         }
 
+        self.deliver_shell_completions("").await;
+
         // A successful tool step is productive progress, not a runaway
         // synthetic resume. Declared per-task tool budgets and max_steps
         // remain the explicit limits for tool-driven work.

crates/tui/src/tools/shell.rs

@@ -3576,7 +3576,11 @@ impl ShellManager {
             let owned = active_session_id.is_none_or(|session_id| {
                 !session_id.is_empty() && shell.owner_session_id == session_id
             });
-            if owned && shell.status != ShellStatus::Running && !shell.completion_reported {
+            if owned
+                && shell.owner_agent.is_none()
+                && shell.status != ShellStatus::Running
+                && !shell.completion_reported
+            {
                 shell.completion_reported = true;
                 completions.push(shell.completion_evidence());
                 // The bytes are now in the caller's hands (they become a durable
@@ -5384,8 +5388,13 @@ fn finish_contract_bash_result(
             tail_lines(output.trim(), 20)
         };
         let notice = promotion_notice(context, task_id, result.duration_ms / 1_000, &output);
+        let completion = if context.owner_agent_id.is_some() {
+            "Its completion stays in task/status and is not injected into the parent model."
+        } else {
+            "Its completion reaches you as a runtime event before your next step; if you end the turn first, a headless run stops it unseen."
+        };
         return Ok(ToolResult::success(format!(
-            "{notice}\n\nOutput so far:\n{so_far}\n\nCompletion will appear as a runtime event. To see more output or block until it finishes, call `task_shell_wait` directly with task_id=\"{task_id}\"; do not sleep-poll with bash. To stop it, send the same call with cancel=true. It also stops when the session ends."
+            "{notice}\n\nOutput so far:\n{so_far}\n\n{completion} To see more output or block until it finishes, call `task_shell_wait` directly with task_id=\"{task_id}\"; do not sleep-poll with bash. To stop it, send the same call with cancel=true. It also stops when the session ends."
         )).with_metadata(metadata));
     }
     if result.status != ShellStatus::Completed {
@@ -6491,7 +6500,7 @@ impl ToolSpec for BashTool {
                     let completion_contract = if context.owner_agent_id.is_some() {
                         "completion stays in task/status and is not injected into the parent model."
                     } else {
-                        "completion is delivered to the model as an internal runtime event and shown in task/status state."
+                        "completion is delivered to the model as an internal runtime event before its next step and shown in task/status state; if the turn ends first, a headless run stops the task unseen."
                     };
                     if backgrounded_foreground {
                         let seconds = result.duration_ms / 1_000;