feat(delegate): warn the child at 80% of its inactivity window before abandoning it

A child that stalls under a configured delegation.child_timeout_seconds used to
learn about the budget only by dying, losing its whole context. The liveness
wait now queues a one-line "[delegation budget warning]" through the child's
steer channel once the idle window is 80% spent (delivered at the child's next
iteration boundary), so a slow-but-recoverable child can wrap up and return its
summary. The warning fires once per idle window and re-arms when progress
resets the window; a progressing child never sees it.

Part of #116001 (atom 2A). Semantics of child_timeout_seconds are unchanged.

tests/tools/test_delegate_liveness_timeout.py

@@ -42,6 +42,11 @@ class _SlowButLiveChild:
         self._calls = initial_calls
         self._activity_ts = time.time()
         self.interrupted = threading.Event()
+        self.steers: list[tuple[float, str]] = []  # (monotonic time, text) of every budget warning received
+
+    def steer(self, text: str) -> bool:
+        self.steers.append((time.monotonic(), text))
+        return True
 
     def run_conversation(self, **_kwargs):
         deadline = time.monotonic() + self._total_seconds
@@ -92,6 +97,8 @@ def test_progressing_child_outlives_a_cap_shorter_than_its_runtime(monkeypatch):
     # The cap really was armed and the child really did run past it: the budget reset on progress.
     assert entry["duration_seconds"] > _CAP_SECONDS, entry
     assert not child.interrupted.is_set()
+    # Progress kept resetting the window, so the 80% budget warning never had cause to fire.
+    assert child.steers == [], child.steers
 
 
 def test_frozen_child_is_still_abandoned_when_the_cap_elapses(monkeypatch):
@@ -105,3 +112,18 @@ def test_frozen_child_is_still_abandoned_when_the_cap_elapses(monkeypatch):
     assert entry["timeout_seconds"] == _CAP_SECONDS
     assert entry["last_event_age"] is not None and entry["last_event_age"] > 0.3, entry
     assert child.interrupted.is_set()
+
+
+def test_frozen_child_is_warned_once_at_80_percent_of_the_window(monkeypatch):
+    """Atom 2A of #116001: a stalling child hears about the closing window while it can still wrap up."""
+    child = _SlowButLiveChild(total_seconds=1.2, advance=False, initial_calls=3)
+    started = time.monotonic()
+
+    entry = _run(child, monkeypatch)
+
+    assert entry["status"] == "timeout", entry
+    assert len(child.steers) == 1, child.steers
+    warned_at, text = child.steers[0]
+    assert "[delegation budget warning]" in text and f"{_CAP_SECONDS:.0f}s inactivity window" in text, text
+    # Fired inside the window (after ~80% of it, before the kill), not at the timeout itself.
+    assert 0.8 * _CAP_SECONDS - 0.05 <= warned_at - started < _CAP_SECONDS, (warned_at - started, _CAP_SECONDS)

tools/delegate_tool_child_run.py

@@ -224,6 +224,26 @@ def _dump_subagent_timeout_diagnostic(
 # Granularity for re-checking a child's progress while a configured ``child_timeout_seconds`` budget runs.
 # Five seconds is finer than the 30s heartbeat and cheap (one activity-summary read per slice).
 _LIVENESS_POLL_SECONDS = 5.0
+# Fraction of the inactivity budget after which the child is warned once (via its steer channel) that the window is
+# closing, so a child that is merely slow can wrap up and return instead of losing its whole context (#116001).
+_BUDGET_WARNING_FRACTION = 0.8
+
+def _budget_warning_text(idle_seconds: float, child_timeout: float) -> str:
+    return (
+        f"[delegation budget warning] No progress signal for {idle_seconds:.0f}s of your {child_timeout:.0f}s "
+        "inactivity window. Finish the current step and return your summary now — the work is discarded if the "
+        "window elapses."
+    )
+
+def _warn_child_budget(child: Any, idle_seconds: float, child_timeout: float) -> None:
+    """Queue the one-line warning through the child's steer path (delivered at its next iteration boundary)."""
+    steer = getattr(child, "steer", None)
+    if not callable(steer):
+        return
+    try:
+        steer(_budget_warning_text(idle_seconds, child_timeout))
+    except Exception as exc:
+        logger.debug("budget warning steer failed: %s", exc)
 
 def _child_activity_fingerprint(child: Any) -> tuple:
     """``(completed calls, current tool, activity clock)`` — the progress signals the heartbeat's stale verdict
@@ -781,18 +801,29 @@ class _ChildRun:
             settled.wait()
             return
         deadline = time.monotonic() + child_timeout
+        warn_at = deadline - child_timeout * (1.0 - _BUDGET_WARNING_FRACTION)
+        warned = False
         fingerprint = _child_activity_fingerprint(self.child)
         while True:
-            remaining = deadline - time.monotonic()
+            now = time.monotonic()
+            remaining = deadline - now
             if remaining <= 0:
                 return  # no progress for the whole budget
-            settled.wait(timeout=min(_LIVENESS_POLL_SECONDS, remaining))
+            wait = min(_LIVENESS_POLL_SECONDS, remaining)
+            if not warned:
+                wait = min(wait, max(warn_at - now, 0.0))
+            settled.wait(timeout=wait)
             if settled.is_set():
                 return  # the worker finished, or the heartbeat declared the child stale
             current = _child_activity_fingerprint(self.child)
             if current != fingerprint:
                 fingerprint = current
                 deadline = time.monotonic() + child_timeout
+                warn_at = deadline - child_timeout * (1.0 - _BUDGET_WARNING_FRACTION)
+                warned = False  # a fresh window gets its own warning
+            elif not warned and time.monotonic() >= warn_at:
+                warned = True
+                _warn_child_budget(self.child, child_timeout - (deadline - time.monotonic()), child_timeout)
 
     def await_child(self) -> tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]], bool]:
         """Run the child's conversation on a daemon worker: ``(result, None, False)`` or ``(None, error_entry,

website/docs/user-guide/features/delegation.md

@@ -367,6 +367,8 @@ delegation:
 
 A positive value bounds **inactivity, not total runtime**: it is the longest a child may go with *no* progress (no completed API call, no tool change, no activity-clock tick) before it is abandoned. Every sign of progress restarts the window, so a child waiting on a multi-minute completion — the case that used to lose finished work — is never killed for taking long, while a child that has genuinely stopped moving is still caught (and an in-flight request is bounded independently by the per-call stale watchdog). `0` or a negative value disables the cap; the heartbeat staleness monitor below stays active either way.
 
+At ~80% of an idle window the child receives a one-line `[delegation budget warning]` through its steer channel (delivered at its next iteration boundary) telling it how long it has been idle and to return its summary now, so a slow-but-recoverable child can wrap up instead of losing its context. The warning fires once per idle window and re-arms when progress resumes.
+
 When a configured cap or the stale threshold fires, the child's result carries
 structured timeout metadata alongside the error message so parents and hooks
 can distinguish a stopwatch kill from other failures without parsing text: