From d50914319704f80348103af8b0247030d93b52aa Mon Sep 17 00:00:00 2001
From: Anthony Volk <14987227+anth-volk@users.noreply.github.com>
Date: Tue, 13 Oct 2026 23:16:28 +0400
Subject: [PATCH 1/3] Respawn overdue segmented national children and fail
 before the parent timeout

A segmented US national run waited on each regional child with no deadline
of its own. When Modal kept re-running a child whose containers were
terminated, the parent waited until its 3,600 s Modal timeout killed it,
which also left the children running with nobody to cancel them.

On 9 October 2026 both 2023 national reports (7fa9408d and 765eefcf) hit
this: their 20 children were re-executed 180 and 131 times, none finished,
and the parents ended at 3,599.8 s (timeout) and 3,167 s ("Server has lost
track of input" from a child poll).

The runner now:
- cancels a child with no result 900 s after dispatch and respawns its
  region group once, recording a segmented_national_child_respawned event
  and segmented_child_retry_count;
- fails with a message naming the group when the respawn is also overdue,
  fails, or the parent budget is too short to fit a retry;
- fails in-band and cancels every child when the parent budget (the
  3,600 s function timeout minus a 300 s completion margin) runs out.

A unit test pins the budget constant to the run_simulation timeout in
src/modal/app.py.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/modal/segmented_national.py           | 227 +++++++++++++++---
 .../tests/test_canonical_spm.py               |  11 +-
 .../tests/test_segmented_national.py          | 196 ++++++++++++++-
 3 files changed, 394 insertions(+), 40 deletions(-)

diff --git a/projects/policyengine-simulation-executor/src/modal/segmented_national.py b/projects/policyengine-simulation-executor/src/modal/segmented_national.py
index 9e5ed5cc2..d09b9167a 100644
--- a/projects/policyengine-simulation-executor/src/modal/segmented_national.py
+++ b/projects/policyengine-simulation-executor/src/modal/segmented_national.py
@@ -15,13 +15,18 @@
 There is no batch state store: the plain job-poll path needs no partial
 status, so a parent failure propagates to the gateway as a failed job. A
 parent killed outright (its own timeout/OOM) cannot cancel children — the
-in-band failure paths do.
+in-band failure paths do. The parent therefore enforces its own deadlines
+below the Modal timeout: a child with no result within its deadline is
+cancelled and respawned once, and the run fails in-band (cancelling every
+child) before the parent's Modal timeout would kill it.
 """
 
 from __future__ import annotations
 
 import logging
 import time
+from collections.abc import Callable
+from dataclasses import dataclass
 from typing import Any
 
 from policyengine_observability import ObservabilityRuntime
@@ -55,6 +60,21 @@
 # The dedicated children pool (see module docstring).
 SEGMENT_FUNCTION_NAME = "run_simulation_segment"
 
+# Seconds from dispatch before a child with no result is cancelled and
+# respawned. Successful children took 124 s at the median and at most 302 s
+# in the week to 13 October 2026, apart from one 2,672 s outlier.
+CHILD_DEADLINE_SECONDS = 900.0
+# Must equal the run_simulation timeout in src/modal/app.py.
+PARENT_FUNCTION_TIMEOUT_SECONDS = 3600.0
+# Parent time kept back for the reduce, result return, and the work done
+# before this runner starts.
+PARENT_COMPLETION_MARGIN_SECONDS = 300.0
+# A respawn needs at least this long to be worth starting (the slowest
+# successful children took about 300 s).
+MINIMUM_RETRY_WINDOW_SECONDS = 300.0
+# The first dispatch plus one respawn after a missed deadline.
+MAX_CHILD_ATTEMPTS = 2
+
 # Parent-scoped keys never forwarded to children. (The shared payload
 # builder owns _telemetry: stripped, then re-attached when present.)
 # `region` is stripped because v1-style national parents carry
@@ -133,6 +153,21 @@ def build_group_child_payload(
     )
 
 
+@dataclass
+class ChildExecution:
+    """One dispatched attempt at a region group's child simulation."""
+
+    group: list[str]
+    call: Any
+    attempt: int
+    dispatched_at: float
+    deadline_seconds: float
+
+    @property
+    def group_label(self) -> str:
+        return "+".join(self.group)
+
+
 class SegmentedNationalRunner:
     """Runs one segmented national request to completion."""
 
@@ -145,6 +180,11 @@ def __init__(
         modal_module=None,
         poll_interval_seconds: float = POLL_INTERVAL_INITIAL_SECONDS,
         poll_interval_max_seconds: float = POLL_INTERVAL_MAX_SECONDS,
+        child_deadline_seconds: float = CHILD_DEADLINE_SECONDS,
+        parent_budget_seconds: float = (
+            PARENT_FUNCTION_TIMEOUT_SECONDS - PARENT_COMPLETION_MARGIN_SECONDS
+        ),
+        clock: Callable[[], float] = time.monotonic,
     ):
         self.params = params
         self.app_name = app_name
@@ -152,6 +192,13 @@ def __init__(
         self.modal = modal if modal_module is None else modal_module
         self.poll_interval_initial_seconds = poll_interval_seconds
         self.poll_interval_max_seconds = poll_interval_max_seconds
+        self.child_deadline_seconds = child_deadline_seconds
+        self.parent_budget_seconds = parent_budget_seconds
+        self.clock = clock
+        # The parent's Modal timeout started before this runner existed; the
+        # completion margin inside the default budget covers that gap.
+        self.budget_ends_at = clock() + parent_budget_seconds
+        self.child_retry_count = 0
         self.country = str(params.get("country") or "us").lower()
         groups = national_region_groups(self.country)
         if not groups:
@@ -171,8 +218,8 @@ def __init__(
     def run(self) -> dict[str, Any]:
         country_module = self._country_module()
         self._extend_last_group_with_uncovered_states(country_module)
-        handles = self._spawn_all_children()
-        child_results = self._collect(handles)
+        children = self._spawn_all_children()
+        child_results = self._collect(children)
         return self._reduce(child_results, country_module=country_module)
 
     def _country_module(self):
@@ -209,33 +256,42 @@ def _extend_last_group_with_uncovered_states(self, country_module) -> None:
             self.runtime.set_context(partition_uncovered_regions=",".join(uncovered))
             self.groups[-1] = list(self.groups[-1]) + uncovered
 
-    def _spawn_all_children(self) -> list[tuple[list[str], Any]]:
-        handles: list[tuple[list[str], Any]] = []
+    def _spawn_child(self, group: list[str], *, attempt: int) -> Any:
+        payload = build_group_child_payload(self.params, group)
+        observability_context = self.runtime.capture_context()
+        with self.runtime.span(
+            SEGMENTED_NATIONAL_STAGES.name(Stage.SEGMENTED_NATIONAL_CHILD_SPAWN),
+            attributes={"region_group": "+".join(group), "child_attempt": attempt},
+        ):
+            return self.child_func.spawn(
+                payload,
+                observability_context=observability_context,
+            )
+
+    def _spawn_all_children(self) -> list[ChildExecution]:
+        children: list[ChildExecution] = []
         try:
             for group in self.groups:
-                payload = build_group_child_payload(self.params, group)
-                observability_context = self.runtime.capture_context()
-                with self.runtime.span(
-                    SEGMENTED_NATIONAL_STAGES.name(
-                        Stage.SEGMENTED_NATIONAL_CHILD_SPAWN
-                    ),
-                    attributes={"region_group": "+".join(group)},
-                ):
-                    call = self.child_func.spawn(
-                        payload,
-                        observability_context=observability_context,
+                call = self._spawn_child(group, attempt=1)
+                children.append(
+                    ChildExecution(
+                        group=group,
+                        call=call,
+                        attempt=1,
+                        dispatched_at=self.clock(),
+                        deadline_seconds=self.child_deadline_seconds,
                     )
-                handles.append((group, call))
+                )
         except Exception:
             # Children spawned before the failure must not run for a job
             # that is already dead.
-            self._cancel_all(handles)
+            self._cancel_all(children)
             raise
-        return handles
+        return children
 
-    def _collect(self, handles: list[tuple[list[str], Any]]) -> list[dict]:
-        results: list[dict | None] = [None] * len(handles)
-        pending = set(range(len(handles)))
+    def _collect(self, children: list[ChildExecution]) -> list[dict]:
+        results: list[dict | None] = [None] * len(children)
+        pending = set(range(len(children)))
         poll_errors: dict[int, int] = {}
         current_sleep = self.poll_interval_initial_seconds
         poll_count = 0
@@ -243,19 +299,24 @@ def _collect(self, handles: list[tuple[list[str], Any]]) -> list[dict]:
         while pending:
             progress_made = False
             for index in sorted(pending):
-                group, call = handles[index]
+                child = children[index]
                 poll_count += 1
                 try:
-                    results[index] = call.get(timeout=0)
+                    results[index] = child.call.get(timeout=0)
                 except TimeoutError:
                     # A not-ready probe is a SUCCESSFUL RPC: reset the error
                     # tally so only consecutive failures fail the job.
                     poll_errors.pop(index, None)
+                    now = self.clock()
+                    if now - child.dispatched_at >= child.deadline_seconds:
+                        children[index] = self._respawn_overdue_child(
+                            children, index, now=now
+                        )
                     continue
                 except Exception as exc:
                     detail = spm_error_detail(exc)
                     if detail:
-                        self._cancel_all(handles)
+                        self._cancel_all(children)
                         raise SPMInputError(detail.code, detail.message) from exc
                     # One transient poll-RPC blip must not kill the job; a
                     # real child failure re-raises on the next probe too.
@@ -264,7 +325,7 @@ def _collect(self, handles: list[tuple[list[str], Any]]) -> list[dict]:
                     if attempts < 2:
                         logger.warning(
                             "Poll error for group %s (attempt %d), retrying: %s",
-                            "+".join(group),
+                            child.group_label,
                             attempts,
                             type(exc).__name__,
                         )
@@ -273,12 +334,20 @@ def _collect(self, handles: list[tuple[list[str], Any]]) -> list[dict]:
                         exc,
                         runtime=self.runtime,
                         scope="segmented_national_child",
-                        context={"region_group": "+".join(group)},
+                        context={
+                            "region_group": child.group_label,
+                            "child_attempt": child.attempt,
+                        },
+                    )
+                    self._cancel_all(children)
+                    retry_note = (
+                        f" on attempt {child.attempt}, after a deadline respawn"
+                        if child.attempt > 1
+                        else ""
                     )
-                    self._cancel_all(handles)
                     raise RuntimeError(
                         "Segmented national child failed for group "
-                        f"{'+'.join(group)}: {redacted}"
+                        f"{child.group_label}{retry_note}: {redacted}"
                     ) from None
                 pending.discard(index)
                 poll_errors.pop(index, None)
@@ -286,6 +355,7 @@ def _collect(self, handles: list[tuple[list[str], Any]]) -> list[dict]:
             # Bounded aggregate instead of one segment per probe (see the
             # Stage registry note about long poll loops).
             self.runtime.set_context(segmented_poll_count=poll_count)
+            self._fail_if_budget_exhausted(children, pending)
             if pending and not progress_made:
                 time.sleep(current_sleep)
                 current_sleep = next_backoff(
@@ -297,13 +367,100 @@ def _collect(self, handles: list[tuple[list[str], Any]]) -> list[dict]:
                 current_sleep = self.poll_interval_initial_seconds
         return results  # type: ignore[return-value]
 
-    def _cancel_all(self, handles: list[tuple[list[str], Any]]) -> None:
+    def _respawn_overdue_child(
+        self, children: list[ChildExecution], index: int, *, now: float
+    ) -> ChildExecution:
+        """Replace a child that missed its deadline, or fail the run.
+
+        Modal can keep re-running a child whose containers are terminated, so
+        a stuck child never fails on its own; only the parent can stop it.
+        """
+        overdue = children[index]
+        waited_seconds = now - overdue.dispatched_at
+        remaining_budget_seconds = self.budget_ends_at - now
+        if overdue.attempt >= MAX_CHILD_ATTEMPTS:
+            reason = (
+                f"no result {waited_seconds:.0f}s after dispatch on attempt "
+                f"{overdue.attempt} of {MAX_CHILD_ATTEMPTS}"
+            )
+        elif remaining_budget_seconds < MINIMUM_RETRY_WINDOW_SECONDS:
+            reason = (
+                f"no result {waited_seconds:.0f}s after dispatch, and only "
+                f"{max(remaining_budget_seconds, 0):.0f}s of the parent budget "
+                "remains for a retry"
+            )
+        else:
+            reason = None
+        if reason is not None:
+            self._cancel_all(children)
+            raise RuntimeError(
+                "Segmented national child timed out for group "
+                f"{overdue.group_label}: {reason}"
+            ) from None
+
+        self._cancel_call(overdue.call)
+        next_attempt = overdue.attempt + 1
+        self.child_retry_count += 1
+        logger.warning(
+            "Group %s had no result %.0fs after dispatch; respawning "
+            "(attempt %d of %d)",
+            overdue.group_label,
+            waited_seconds,
+            next_attempt,
+            MAX_CHILD_ATTEMPTS,
+        )
+        self.runtime.event(
+            "segmented_national_child_respawned",
+            severity="WARNING",
+            attributes={
+                "region_group": overdue.group_label,
+                "child_attempt": next_attempt,
+                "waited_seconds": round(waited_seconds, 3),
+                "deadline_seconds": overdue.deadline_seconds,
+            },
+        )
+        self.runtime.set_context(segmented_child_retry_count=self.child_retry_count)
+        try:
+            call = self._spawn_child(overdue.group, attempt=next_attempt)
+        except Exception:
+            self._cancel_all(children)
+            raise
+        return ChildExecution(
+            group=overdue.group,
+            call=call,
+            attempt=next_attempt,
+            dispatched_at=self.clock(),
+            deadline_seconds=min(self.child_deadline_seconds, remaining_budget_seconds),
+        )
+
+    def _fail_if_budget_exhausted(
+        self, children: list[ChildExecution], pending: set[int]
+    ) -> None:
+        """Fail in-band, cancelling children, before the Modal timeout can
+        kill the parent and strand them."""
+        if not pending or self.clock() < self.budget_ends_at:
+            return
+        self._cancel_all(children)
+        outstanding_groups = ", ".join(
+            children[index].group_label for index in sorted(pending)
+        )
+        raise RuntimeError(
+            "Segmented national run exceeded its "
+            f"{self.parent_budget_seconds:.0f}s budget with groups still "
+            f"running: {outstanding_groups}"
+        )
+
+    @staticmethod
+    def _cancel_call(call: Any) -> None:
+        try:
+            call.cancel()
+        except Exception:
+            pass
+
+    def _cancel_all(self, children: list[ChildExecution]) -> None:
         """Best-effort cancel so a failed batch doesn't strand workers."""
-        for _, call in handles:
-            try:
-                call.cancel()
-            except Exception:
-                pass
+        for child in children:
+            self._cancel_call(child.call)
 
     def _reduce(self, child_results: list[dict], *, country_module) -> dict[str, Any]:
         from policyengine_simulation_executor.simulation_runtime import (
diff --git a/projects/policyengine-simulation-executor/tests/test_canonical_spm.py b/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
index b091a9a23..d331784d2 100644
--- a/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
+++ b/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
@@ -739,7 +739,7 @@ def test_gateway_submission_uses_registry_capability_before_spawn(monkeypatch):
 
 
 def test_segmented_child_input_error_is_not_redacted_or_retried(monkeypatch):
-    from src.modal.segmented_national import SegmentedNationalRunner
+    from src.modal.segmented_national import ChildExecution, SegmentedNationalRunner
 
     child = SimpleNamespace(
         get=Mock(
@@ -747,11 +747,18 @@ def test_segmented_child_input_error_is_not_redacted_or_retried(monkeypatch):
         ),
         cancel=Mock(),
     )
+    child_execution = ChildExecution(
+        group=["state/ca"],
+        call=child,
+        attempt=1,
+        dispatched_at=0.0,
+        deadline_seconds=900.0,
+    )
     runner = object.__new__(SegmentedNationalRunner)
     runner.poll_interval_initial_seconds = 0.001
     runner.poll_interval_max_seconds = 0.001
     with pytest.raises(SPMInputError) as error:
-        runner._collect([(["state/ca"], child)])
+        runner._collect([child_execution])
     assert error.value.code == "SPM_GEOGRAPHY_REQUIRED"
     child.get.assert_called_once()
     child.cancel.assert_called_once()
diff --git a/projects/policyengine-simulation-executor/tests/test_segmented_national.py b/projects/policyengine-simulation-executor/tests/test_segmented_national.py
index 547c202e3..f4c2e820c 100644
--- a/projects/policyengine-simulation-executor/tests/test_segmented_national.py
+++ b/projects/policyengine-simulation-executor/tests/test_segmented_national.py
@@ -1,5 +1,7 @@
 """Unit tests for the segmented-national fan-out runner (C5/C6 + review fixes)."""
 
+import ast
+from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
@@ -141,14 +143,24 @@ def test__strips_v1_style_region_us_from_children(self):
 class FakeCall:
     """A FunctionCall whose result resolves after N polls (or raises)."""
 
-    def __init__(self, result, polls_until_ready=0, error=None, errors_once=0):
+    def __init__(
+        self,
+        result,
+        polls_until_ready=0,
+        error=None,
+        errors_once=0,
+        before_each_poll=None,
+    ):
         self.result = result
         self.polls_until_ready = polls_until_ready
         self.error = error
         self.errors_once = errors_once
+        self.before_each_poll = before_each_poll
         self.cancelled = False
 
     def get(self, timeout):
+        if self.before_each_poll is not None:
+            self.before_each_poll()
         if self.errors_once > 0:
             self.errors_once -= 1
             raise ConnectionError("transient control-plane blip")
@@ -191,14 +203,15 @@ def _spawn(self, payload, *, observability_context=None):
         return self._calls[len(self.spawned_payloads) - 1]
 
 
-def _runner(fake_modal, params=None):
+def _runner(fake_modal, params=None, runtime=TEST_RUNTIME, **runner_options):
     return sn.SegmentedNationalRunner(
         params or dict(NATIONAL),
         app_name="policyengine-simulation-py-test",
-        runtime=TEST_RUNTIME,
+        runtime=runtime,
         modal_module=fake_modal,
         poll_interval_seconds=0.001,
         poll_interval_max_seconds=0.002,
+        **runner_options,
     )
 
 
@@ -345,6 +358,183 @@ def test__uk_has_no_partition_to_run(self):
             _runner(FakeModal([]), params={"country": "uk", "scope": "macro"})
 
 
+class ManualClock:
+    """A monotonic clock that only moves when a test advances it."""
+
+    def __init__(self):
+        self.now_seconds = 0.0
+
+    def __call__(self):
+        return self.now_seconds
+
+    def advance(self, seconds):
+        self.now_seconds += seconds
+
+
+class RecordingRuntime(NoOpObservabilityRuntime):
+    def __init__(self):
+        self.events = []
+        self.context_updates = []
+
+    def event(self, name, **kwargs):
+        self.events.append((name, kwargs))
+
+    def set_context(self, **attributes):
+        self.context_updates.append(attributes)
+
+
+STUCK_GROUP_INDEX = 4
+SECONDS_PER_STUCK_POLL = 100
+
+
+def _stuck_call(clock):
+    """A child that never finishes; each poll of it moves the clock on."""
+    return FakeCall(
+        None,
+        polls_until_ready=10**6,
+        before_each_poll=lambda: clock.advance(SECONDS_PER_STUCK_POLL),
+    )
+
+
+def _calls_with_stuck_child(clock, respawned_call):
+    calls = _twenty_calls()
+    calls[STUCK_GROUP_INDEX] = _stuck_call(clock)
+    return [*calls, respawned_call]
+
+
+class TestSegmentedNationalChildDeadline:
+    def test__overdue_child_is_cancelled_and_respawned_once(self, bare_country):
+        clock = ManualClock()
+        respawned_call = FakeCall({"child": "respawned"})
+        calls = _calls_with_stuck_child(clock, respawned_call)
+        fake = FakeModal(calls)
+        runner = _runner(fake, clock=clock)
+        runner._reduce = lambda results, country_module: results
+
+        results = runner.run()
+
+        expected_results = [{"child": index} for index in range(20)]
+        expected_results[STUCK_GROUP_INDEX] = {"child": "respawned"}
+        assert results == expected_results
+        assert calls[STUCK_GROUP_INDEX].cancelled
+        assert len(fake.spawned_payloads) == 21
+        assert fake.spawned_payloads[-1]["region_group"] == list(
+            runner.groups[STUCK_GROUP_INDEX]
+        )
+        assert not any(
+            call.cancelled
+            for index, call in enumerate(calls)
+            if index != STUCK_GROUP_INDEX
+        )
+        # The respawn waited for the full default deadline, not longer.
+        assert clock() == sn.CHILD_DEADLINE_SECONDS
+
+    def test__respawned_child_that_is_also_overdue_fails_the_run(self, bare_country):
+        clock = ManualClock()
+        calls = _calls_with_stuck_child(clock, _stuck_call(clock))
+        runner = _runner(FakeModal(calls), clock=clock)
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"timed out for group .*on attempt 2 of 2",
+        ) as raised:
+            runner.run()
+
+        assert "+".join(runner.groups[STUCK_GROUP_INDEX]) in str(raised.value)
+        assert calls[STUCK_GROUP_INDEX].cancelled
+        assert calls[-1].cancelled
+
+    def test__respawned_child_failure_names_the_respawn(self, bare_country):
+        clock = ManualClock()
+        failing_respawn = FakeCall(None, error=ValueError("child crashed"))
+        calls = _calls_with_stuck_child(clock, failing_respawn)
+        runner = _runner(FakeModal(calls), clock=clock)
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"child failed for group .* on attempt 2, after a deadline respawn",
+        ):
+            runner.run()
+        assert calls[STUCK_GROUP_INDEX].cancelled
+
+    def test__no_respawn_when_parent_budget_cannot_fit_a_retry(self, bare_country):
+        clock = ManualClock()
+        calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
+        fake = FakeModal(calls)
+        # At the 900 s child deadline only 100 s of a 1,000 s budget remain.
+        runner = _runner(fake, clock=clock, parent_budget_seconds=1000)
+
+        with pytest.raises(
+            RuntimeError, match=r"only 100s of the parent budget remains"
+        ):
+            runner.run()
+
+        assert len(fake.spawned_payloads) == 20
+        assert calls[STUCK_GROUP_INDEX].cancelled
+
+    def test__exhausted_parent_budget_cancels_children_before_modal_timeout(
+        self, bare_country
+    ):
+        clock = ManualClock()
+        calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
+        fake = FakeModal(calls)
+        # The budget runs out before the child's own deadline does.
+        runner = _runner(
+            fake,
+            clock=clock,
+            parent_budget_seconds=500,
+            child_deadline_seconds=10_000,
+        )
+
+        with pytest.raises(
+            RuntimeError,
+            match=r"exceeded its 500s budget with groups still running",
+        ) as raised:
+            runner.run()
+
+        assert "+".join(runner.groups[STUCK_GROUP_INDEX]) in str(raised.value)
+        assert calls[STUCK_GROUP_INDEX].cancelled
+        assert len(fake.spawned_payloads) == 20
+
+    def test__respawn_is_recorded_in_telemetry(self, bare_country):
+        clock = ManualClock()
+        runtime = RecordingRuntime()
+        calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
+        runner = _runner(FakeModal(calls), runtime=runtime, clock=clock)
+        runner._reduce = lambda results, country_module: results
+
+        runner.run()
+
+        respawn_events = [
+            options
+            for name, options in runtime.events
+            if name == "segmented_national_child_respawned"
+        ]
+        assert len(respawn_events) == 1
+        assert respawn_events[0]["severity"] == "WARNING"
+        assert respawn_events[0]["attributes"]["region_group"] == "+".join(
+            runner.groups[STUCK_GROUP_INDEX]
+        )
+        assert respawn_events[0]["attributes"]["child_attempt"] == 2
+        assert {"segmented_child_retry_count": 1} in runtime.context_updates
+
+    def test__parent_budget_matches_the_modal_function_timeout(self):
+        app_source = Path(sn.__file__).with_name("app.py").read_text()
+        run_simulation_definition = next(
+            node
+            for node in ast.walk(ast.parse(app_source))
+            if isinstance(node, ast.FunctionDef) and node.name == "run_simulation"
+        )
+        timeouts = [
+            keyword.value.value
+            for decorator in run_simulation_definition.decorator_list
+            if isinstance(decorator, ast.Call)
+            for keyword in decorator.keywords
+            if keyword.arg == "timeout"
+        ]
+        assert timeouts == [sn.PARENT_FUNCTION_TIMEOUT_SECONDS]
+
+
 class TestDispatchRunSimulation:
     def test__eligible_national_takes_the_segmented_path(self, monkeypatch):
         calls = {}

From 26028c982f8ede5458dc1dcbdaa3b7949aa055a8 Mon Sep 17 00:00:00 2001
From: Anthony Volk <14987227+anth-volk@users.noreply.github.com>
Date: Wed, 14 Oct 2026 18:37:07 +0400
Subject: [PATCH 2/3] Keep cancelling superseded segmented children and record
 timeout failures

A respawned child's original attempt was cancelled once, with errors
ignored, and then dropped from tracking, so later failure paths never
retried the cancel. Superseded attempts now stay tracked: they are never
polled again, but every failure path and the end of collection cancel
them again.

The final deadline failure and parent budget exhaustion now emit
structured events (segmented_national_child_timed_out and
segmented_national_budget_exhausted). The respawn event and
segmented_child_retry_count are recorded only after the respawn spawns.

Adds tests for retrying a failed cancel, ignoring a late result from the
superseded original, a failed respawn spawn, clamping a respawn's
deadline to the remaining budget, and the new events.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/modal/segmented_national.py           |  63 +++++-
 .../tests/test_canonical_spm.py               |   1 +
 .../tests/test_segmented_national.py          | 204 +++++++++++++++++-
 3 files changed, 249 insertions(+), 19 deletions(-)

diff --git a/projects/policyengine-simulation-executor/src/modal/segmented_national.py b/projects/policyengine-simulation-executor/src/modal/segmented_national.py
index d09b9167a..bf45e00a1 100644
--- a/projects/policyengine-simulation-executor/src/modal/segmented_national.py
+++ b/projects/policyengine-simulation-executor/src/modal/segmented_national.py
@@ -197,8 +197,13 @@ def __init__(
         self.clock = clock
         # The parent's Modal timeout started before this runner existed; the
         # completion margin inside the default budget covers that gap.
-        self.budget_ends_at = clock() + parent_budget_seconds
+        self.started_at = clock()
+        self.budget_ends_at = self.started_at + parent_budget_seconds
         self.child_retry_count = 0
+        # Overdue attempts replaced by a respawn. They are never polled
+        # again, but every later cancel retries them: the first cancel can
+        # fail, and Modal keeps re-running a stuck input until one succeeds.
+        self.superseded_children: list[ChildExecution] = []
         self.country = str(params.get("country") or "us").lower()
         groups = national_region_groups(self.country)
         if not groups:
@@ -365,6 +370,9 @@ def _collect(self, children: list[ChildExecution]) -> list[dict]:
                 )
             elif progress_made:
                 current_sleep = self.poll_interval_initial_seconds
+        # Every group has a result; stop any superseded attempt whose first
+        # cancel failed.
+        self._cancel_superseded_children()
         return results  # type: ignore[return-value]
 
     def _respawn_overdue_child(
@@ -392,6 +400,19 @@ def _respawn_overdue_child(
         else:
             reason = None
         if reason is not None:
+            self.runtime.event(
+                "segmented_national_child_timed_out",
+                severity="ERROR",
+                attributes={
+                    "region_group": overdue.group_label,
+                    "child_attempt": overdue.attempt,
+                    "waited_seconds": round(waited_seconds, 3),
+                    "deadline_seconds": overdue.deadline_seconds,
+                    "remaining_budget_seconds": round(
+                        max(remaining_budget_seconds, 0), 3
+                    ),
+                },
+            )
             self._cancel_all(children)
             raise RuntimeError(
                 "Segmented national child timed out for group "
@@ -399,10 +420,16 @@ def _respawn_overdue_child(
             ) from None
 
         self._cancel_call(overdue.call)
+        self.superseded_children.append(overdue)
         next_attempt = overdue.attempt + 1
+        try:
+            call = self._spawn_child(overdue.group, attempt=next_attempt)
+        except Exception:
+            self._cancel_all(children)
+            raise
         self.child_retry_count += 1
         logger.warning(
-            "Group %s had no result %.0fs after dispatch; respawning "
+            "Group %s had no result %.0fs after dispatch; respawned it "
             "(attempt %d of %d)",
             overdue.group_label,
             waited_seconds,
@@ -420,11 +447,6 @@ def _respawn_overdue_child(
             },
         )
         self.runtime.set_context(segmented_child_retry_count=self.child_retry_count)
-        try:
-            call = self._spawn_child(overdue.group, attempt=next_attempt)
-        except Exception:
-            self._cancel_all(children)
-            raise
         return ChildExecution(
             group=overdue.group,
             call=call,
@@ -438,12 +460,26 @@ def _fail_if_budget_exhausted(
     ) -> None:
         """Fail in-band, cancelling children, before the Modal timeout can
         kill the parent and strand them."""
-        if not pending or self.clock() < self.budget_ends_at:
+        if not pending:
             return
-        self._cancel_all(children)
-        outstanding_groups = ", ".join(
-            children[index].group_label for index in sorted(pending)
+        now = self.clock()
+        if now < self.budget_ends_at:
+            return
+        outstanding = [children[index] for index in sorted(pending)]
+        outstanding_groups = ", ".join(child.group_label for child in outstanding)
+        self.runtime.event(
+            "segmented_national_budget_exhausted",
+            severity="ERROR",
+            attributes={
+                "region_groups": outstanding_groups,
+                "child_attempts": ", ".join(
+                    str(child.attempt) for child in outstanding
+                ),
+                "waited_seconds": round(now - self.started_at, 3),
+                "parent_budget_seconds": self.parent_budget_seconds,
+            },
         )
+        self._cancel_all(children)
         raise RuntimeError(
             "Segmented national run exceeded its "
             f"{self.parent_budget_seconds:.0f}s budget with groups still "
@@ -457,10 +493,15 @@ def _cancel_call(call: Any) -> None:
         except Exception:
             pass
 
+    def _cancel_superseded_children(self) -> None:
+        for superseded in self.superseded_children:
+            self._cancel_call(superseded.call)
+
     def _cancel_all(self, children: list[ChildExecution]) -> None:
         """Best-effort cancel so a failed batch doesn't strand workers."""
         for child in children:
             self._cancel_call(child.call)
+        self._cancel_superseded_children()
 
     def _reduce(self, child_results: list[dict], *, country_module) -> dict[str, Any]:
         from policyengine_simulation_executor.simulation_runtime import (
diff --git a/projects/policyengine-simulation-executor/tests/test_canonical_spm.py b/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
index d331784d2..7265c97d6 100644
--- a/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
+++ b/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
@@ -757,6 +757,7 @@ def test_segmented_child_input_error_is_not_redacted_or_retried(monkeypatch):
     runner = object.__new__(SegmentedNationalRunner)
     runner.poll_interval_initial_seconds = 0.001
     runner.poll_interval_max_seconds = 0.001
+    runner.superseded_children = []
     with pytest.raises(SPMInputError) as error:
         runner._collect([child_execution])
     assert error.value.code == "SPM_GEOGRAPHY_REQUIRED"
diff --git a/projects/policyengine-simulation-executor/tests/test_segmented_national.py b/projects/policyengine-simulation-executor/tests/test_segmented_national.py
index f4c2e820c..18687a15a 100644
--- a/projects/policyengine-simulation-executor/tests/test_segmented_national.py
+++ b/projects/policyengine-simulation-executor/tests/test_segmented_national.py
@@ -150,15 +150,20 @@ def __init__(
         error=None,
         errors_once=0,
         before_each_poll=None,
+        cancel_failures=0,
     ):
         self.result = result
         self.polls_until_ready = polls_until_ready
         self.error = error
         self.errors_once = errors_once
         self.before_each_poll = before_each_poll
+        self.cancel_failures = cancel_failures
         self.cancelled = False
+        self.cancel_attempts = 0
+        self.poll_count = 0
 
     def get(self, timeout):
+        self.poll_count += 1
         if self.before_each_poll is not None:
             self.before_each_poll()
         if self.errors_once > 0:
@@ -172,6 +177,10 @@ def get(self, timeout):
         return self.result
 
     def cancel(self):
+        self.cancel_attempts += 1
+        if self.cancel_failures > 0:
+            self.cancel_failures -= 1
+            raise ConnectionError("cancel RPC failed")
         self.cancelled = True
 
 
@@ -382,9 +391,14 @@ def event(self, name, **kwargs):
     def set_context(self, **attributes):
         self.context_updates.append(attributes)
 
+    def events_named(self, event_name):
+        return [options for name, options in self.events if name == event_name]
+
 
 STUCK_GROUP_INDEX = 4
 SECONDS_PER_STUCK_POLL = 100
+# A stuck child reaches the default deadline on this poll.
+POLLS_TO_REACH_DEADLINE = int(sn.CHILD_DEADLINE_SECONDS // SECONDS_PER_STUCK_POLL)
 
 
 def _stuck_call(clock):
@@ -396,8 +410,13 @@ def _stuck_call(clock):
     )
 
 
-def _calls_with_stuck_child(clock, respawned_call):
-    calls = _twenty_calls()
+def _crashing_call(clock):
+    """A child that fails outright (takes a clock to match ``_stuck_call``)."""
+    return FakeCall(None, error=ValueError("child crashed"))
+
+
+def _calls_with_stuck_child(clock, respawned_call, **sibling_options):
+    calls = _twenty_calls(**sibling_options)
     calls[STUCK_GROUP_INDEX] = _stuck_call(clock)
     return [*calls, respawned_call]
 
@@ -426,7 +445,8 @@ def test__overdue_child_is_cancelled_and_respawned_once(self, bare_country):
             for index, call in enumerate(calls)
             if index != STUCK_GROUP_INDEX
         )
-        # The respawn waited for the full default deadline, not longer.
+        # The original was respawned as soon as it reached the default
+        # deadline, not later; the respawn finished without moving the clock.
         assert clock() == sn.CHILD_DEADLINE_SECONDS
 
     def test__respawned_child_that_is_also_overdue_fails_the_run(self, bare_country):
@@ -457,6 +477,87 @@ def test__respawned_child_failure_names_the_respawn(self, bare_country):
             runner.run()
         assert calls[STUCK_GROUP_INDEX].cancelled
 
+    def test__failed_respawn_spawn_cancels_every_child(self, bare_country):
+        clock = ManualClock()
+        runtime = RecordingRuntime()
+        calls = _calls_with_stuck_child(
+            clock, FakeCall({"child": "respawned"}), polls_until_ready=10**6
+        )
+        # The 21st spawn, the respawn, fails while every sibling still runs.
+        fake = FakeModal(calls, fail_spawn_at=20)
+        runner = _runner(fake, runtime=runtime, clock=clock)
+
+        with pytest.raises(ConnectionError, match="spawn RPC failed"):
+            runner.run()
+
+        assert all(call.cancelled for call in calls[:20])
+        # A respawn that never spawned is not reported as one.
+        assert not runtime.events_named("segmented_national_child_respawned")
+        assert not any(
+            "segmented_child_retry_count" in update
+            for update in runtime.context_updates
+        )
+
+    def test__late_result_from_the_superseded_child_is_ignored(self, bare_country):
+        clock = ManualClock()
+        # The original's next poll after its deadline would return a result.
+        late_original = FakeCall(
+            {"child": "late original"},
+            polls_until_ready=POLLS_TO_REACH_DEADLINE,
+            before_each_poll=lambda: clock.advance(SECONDS_PER_STUCK_POLL),
+        )
+        calls = _twenty_calls()
+        calls[STUCK_GROUP_INDEX] = late_original
+        calls.append(FakeCall({"child": "respawned"}, polls_until_ready=2))
+        runner = _runner(FakeModal(calls), clock=clock)
+        runner._reduce = lambda results, country_module: results
+
+        results = runner.run()
+
+        assert len(results) == len(runner.groups)
+        assert results[STUCK_GROUP_INDEX] == {"child": "respawned"}
+        assert late_original.poll_count == POLLS_TO_REACH_DEADLINE
+        assert late_original.cancelled
+
+    @pytest.mark.parametrize(
+        "respawned_call_factory",
+        [_crashing_call, _stuck_call],
+        ids=["respawn_fails", "respawn_also_overdue"],
+    )
+    def test__superseded_child_whose_cancel_failed_is_cancelled_on_a_later_failure(
+        self, bare_country, respawned_call_factory
+    ):
+        clock = ManualClock()
+        calls = _calls_with_stuck_child(clock, respawned_call_factory(clock))
+        superseded_original = calls[STUCK_GROUP_INDEX]
+        superseded_original.cancel_failures = 1
+        runner = _runner(FakeModal(calls), clock=clock)
+
+        with pytest.raises(RuntimeError, match="Segmented national child"):
+            runner.run()
+
+        # The cancel at the respawn failed; the later failure path retried it
+        # without polling the superseded original again.
+        assert superseded_original.cancel_attempts == 2
+        assert superseded_original.cancelled
+        assert superseded_original.poll_count == POLLS_TO_REACH_DEADLINE
+
+    def test__superseded_child_whose_cancel_failed_is_cancelled_after_collection(
+        self, bare_country
+    ):
+        clock = ManualClock()
+        calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
+        superseded_original = calls[STUCK_GROUP_INDEX]
+        superseded_original.cancel_failures = 1
+        runner = _runner(FakeModal(calls), clock=clock)
+        runner._reduce = lambda results, country_module: results
+
+        results = runner.run()
+
+        assert results[STUCK_GROUP_INDEX] == {"child": "respawned"}
+        assert superseded_original.cancel_attempts == 2
+        assert superseded_original.cancelled
+
     def test__no_respawn_when_parent_budget_cannot_fit_a_retry(self, bare_country):
         clock = ManualClock()
         calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
@@ -472,6 +573,30 @@ def test__no_respawn_when_parent_budget_cannot_fit_a_retry(self, bare_country):
         assert len(fake.spawned_payloads) == 20
         assert calls[STUCK_GROUP_INDEX].cancelled
 
+    def test__respawn_deadline_is_clamped_to_the_remaining_parent_budget(
+        self, bare_country
+    ):
+        clock = ManualClock()
+        runtime = RecordingRuntime()
+        calls = _calls_with_stuck_child(clock, _stuck_call(clock))
+        # At the 900 s child deadline 600 s of a 1,500 s budget remain: enough
+        # for a respawn, but less than a full child deadline.
+        runner = _runner(
+            FakeModal(calls),
+            runtime=runtime,
+            clock=clock,
+            parent_budget_seconds=1500,
+        )
+
+        with pytest.raises(
+            RuntimeError, match=r"no result 600s after dispatch on attempt 2 of 2"
+        ):
+            runner.run()
+
+        [timeout_event] = runtime.events_named("segmented_national_child_timed_out")
+        assert timeout_event["attributes"]["deadline_seconds"] == 600
+        assert not runtime.events_named("segmented_national_budget_exhausted")
+
     def test__exhausted_parent_budget_cancels_children_before_modal_timeout(
         self, bare_country
     ):
@@ -505,11 +630,7 @@ def test__respawn_is_recorded_in_telemetry(self, bare_country):
 
         runner.run()
 
-        respawn_events = [
-            options
-            for name, options in runtime.events
-            if name == "segmented_national_child_respawned"
-        ]
+        respawn_events = runtime.events_named("segmented_national_child_respawned")
         assert len(respawn_events) == 1
         assert respawn_events[0]["severity"] == "WARNING"
         assert respawn_events[0]["attributes"]["region_group"] == "+".join(
@@ -518,6 +639,73 @@ def test__respawn_is_recorded_in_telemetry(self, bare_country):
         assert respawn_events[0]["attributes"]["child_attempt"] == 2
         assert {"segmented_child_retry_count": 1} in runtime.context_updates
 
+    @pytest.mark.parametrize(
+        "parent_budget_seconds,expected_attempt,expected_remaining_budget_seconds",
+        # A 3,000 s budget fits both 900 s attempts; a 1,000 s budget leaves
+        # too little after the first attempt for a respawn.
+        [(3000, 2, 1200), (1000, 1, 100)],
+        ids=["respawn_also_overdue", "no_budget_for_a_respawn"],
+    )
+    def test__child_timeout_is_recorded_in_telemetry(
+        self,
+        bare_country,
+        parent_budget_seconds,
+        expected_attempt,
+        expected_remaining_budget_seconds,
+    ):
+        clock = ManualClock()
+        runtime = RecordingRuntime()
+        calls = _calls_with_stuck_child(clock, _stuck_call(clock))
+        runner = _runner(
+            FakeModal(calls),
+            runtime=runtime,
+            clock=clock,
+            parent_budget_seconds=parent_budget_seconds,
+        )
+
+        with pytest.raises(RuntimeError, match="timed out for group"):
+            runner.run()
+
+        assert runtime.events_named("segmented_national_child_timed_out") == [
+            {
+                "severity": "ERROR",
+                "attributes": {
+                    "region_group": "+".join(runner.groups[STUCK_GROUP_INDEX]),
+                    "child_attempt": expected_attempt,
+                    "waited_seconds": sn.CHILD_DEADLINE_SECONDS,
+                    "deadline_seconds": sn.CHILD_DEADLINE_SECONDS,
+                    "remaining_budget_seconds": expected_remaining_budget_seconds,
+                },
+            }
+        ]
+
+    def test__exhausted_parent_budget_is_recorded_in_telemetry(self, bare_country):
+        clock = ManualClock()
+        runtime = RecordingRuntime()
+        calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
+        runner = _runner(
+            FakeModal(calls),
+            runtime=runtime,
+            clock=clock,
+            parent_budget_seconds=500,
+            child_deadline_seconds=10_000,
+        )
+
+        with pytest.raises(RuntimeError, match="exceeded its 500s budget"):
+            runner.run()
+
+        assert runtime.events_named("segmented_national_budget_exhausted") == [
+            {
+                "severity": "ERROR",
+                "attributes": {
+                    "region_groups": "+".join(runner.groups[STUCK_GROUP_INDEX]),
+                    "child_attempts": "1",
+                    "waited_seconds": 500,
+                    "parent_budget_seconds": 500,
+                },
+            }
+        ]
+
     def test__parent_budget_matches_the_modal_function_timeout(self):
         app_source = Path(sn.__file__).with_name("app.py").read_text()
         run_simulation_definition = next(

From d50f5bc15bbc7c24074f164b191398b8418cb640 Mon Sep 17 00:00:00 2001
From: Anthony Volk <14987227+anth-volk@users.noreply.github.com>
Date: Thu, 15 Oct 2026 01:43:21 +0400
Subject: [PATCH 3/3] Raise the segmented child deadline to 1,200 s

Give each regional child 1,200 s from dispatch before it is cancelled and
respawned, instead of 900 s. A stuck group now fails after about 2,400 s,
still inside the 3,300 s parent budget, and the respawn still gets a full
deadline. The tests that encoded 900 s budgets now derive them from
CHILD_DEADLINE_SECONDS.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/modal/segmented_national.py           |  2 +-
 .../tests/test_canonical_spm.py               |  8 +++++--
 .../tests/test_segmented_national.py          | 23 ++++++++++++-------
 3 files changed, 22 insertions(+), 11 deletions(-)

diff --git a/projects/policyengine-simulation-executor/src/modal/segmented_national.py b/projects/policyengine-simulation-executor/src/modal/segmented_national.py
index bf45e00a1..b2392d6e8 100644
--- a/projects/policyengine-simulation-executor/src/modal/segmented_national.py
+++ b/projects/policyengine-simulation-executor/src/modal/segmented_national.py
@@ -63,7 +63,7 @@
 # Seconds from dispatch before a child with no result is cancelled and
 # respawned. Successful children took 124 s at the median and at most 302 s
 # in the week to 13 October 2026, apart from one 2,672 s outlier.
-CHILD_DEADLINE_SECONDS = 900.0
+CHILD_DEADLINE_SECONDS = 1200.0
 # Must equal the run_simulation timeout in src/modal/app.py.
 PARENT_FUNCTION_TIMEOUT_SECONDS = 3600.0
 # Parent time kept back for the reduce, result return, and the work done
diff --git a/projects/policyengine-simulation-executor/tests/test_canonical_spm.py b/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
index 7265c97d6..82bfd89bd 100644
--- a/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
+++ b/projects/policyengine-simulation-executor/tests/test_canonical_spm.py
@@ -739,7 +739,11 @@ def test_gateway_submission_uses_registry_capability_before_spawn(monkeypatch):
 
 
 def test_segmented_child_input_error_is_not_redacted_or_retried(monkeypatch):
-    from src.modal.segmented_national import ChildExecution, SegmentedNationalRunner
+    from src.modal.segmented_national import (
+        CHILD_DEADLINE_SECONDS,
+        ChildExecution,
+        SegmentedNationalRunner,
+    )
 
     child = SimpleNamespace(
         get=Mock(
@@ -752,7 +756,7 @@ def test_segmented_child_input_error_is_not_redacted_or_retried(monkeypatch):
         call=child,
         attempt=1,
         dispatched_at=0.0,
-        deadline_seconds=900.0,
+        deadline_seconds=CHILD_DEADLINE_SECONDS,
     )
     runner = object.__new__(SegmentedNationalRunner)
     runner.poll_interval_initial_seconds = 0.001
diff --git a/projects/policyengine-simulation-executor/tests/test_segmented_national.py b/projects/policyengine-simulation-executor/tests/test_segmented_national.py
index 18687a15a..844302540 100644
--- a/projects/policyengine-simulation-executor/tests/test_segmented_national.py
+++ b/projects/policyengine-simulation-executor/tests/test_segmented_national.py
@@ -562,8 +562,12 @@ def test__no_respawn_when_parent_budget_cannot_fit_a_retry(self, bare_country):
         clock = ManualClock()
         calls = _calls_with_stuck_child(clock, FakeCall({"child": "respawned"}))
         fake = FakeModal(calls)
-        # At the 900 s child deadline only 100 s of a 1,000 s budget remain.
-        runner = _runner(fake, clock=clock, parent_budget_seconds=1000)
+        # At the child deadline only 100 s of the parent budget remain.
+        runner = _runner(
+            fake,
+            clock=clock,
+            parent_budget_seconds=sn.CHILD_DEADLINE_SECONDS + 100,
+        )
 
         with pytest.raises(
             RuntimeError, match=r"only 100s of the parent budget remains"
@@ -579,13 +583,13 @@ def test__respawn_deadline_is_clamped_to_the_remaining_parent_budget(
         clock = ManualClock()
         runtime = RecordingRuntime()
         calls = _calls_with_stuck_child(clock, _stuck_call(clock))
-        # At the 900 s child deadline 600 s of a 1,500 s budget remain: enough
-        # for a respawn, but less than a full child deadline.
+        # At the child deadline 600 s of the parent budget remain: enough for
+        # a respawn, but less than a full child deadline.
         runner = _runner(
             FakeModal(calls),
             runtime=runtime,
             clock=clock,
-            parent_budget_seconds=1500,
+            parent_budget_seconds=sn.CHILD_DEADLINE_SECONDS + 600,
         )
 
         with pytest.raises(
@@ -641,9 +645,12 @@ def test__respawn_is_recorded_in_telemetry(self, bare_country):
 
     @pytest.mark.parametrize(
         "parent_budget_seconds,expected_attempt,expected_remaining_budget_seconds",
-        # A 3,000 s budget fits both 900 s attempts; a 1,000 s budget leaves
-        # too little after the first attempt for a respawn.
-        [(3000, 2, 1200), (1000, 1, 100)],
+        # The first budget fits both attempts with 600 s to spare; the second
+        # leaves too little after the first attempt for a respawn.
+        [
+            (2 * sn.CHILD_DEADLINE_SECONDS + 600, 2, 600),
+            (sn.CHILD_DEADLINE_SECONDS + 100, 1, 100),
+        ],
         ids=["respawn_also_overdue", "no_budget_for_a_respawn"],
     )
     def test__child_timeout_is_recorded_in_telemetry(