Coverage for src/footman/_schedule.py: 92%
508 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Build the task DAG and run it — in parallel by default, or sequentially.
3The chain segments plus each task's `pre`/`post` form a dependency graph
4(deduped by task identity). Independent nodes run concurrently on a thread pool;
5a node runs once all its prerequisites have succeeded. footman tasks are almost
6always I/O-bound (they shell out through `footman.run`, releasing the GIL),
7so threads give real concurrency without process isolation.
9Output is buffered per task and flushed atomically on completion, so concurrent
10tasks never interleave.
11"""
13from __future__ import annotations
15import io
16import os
17import sys
18import threading
19import time
20from dataclasses import dataclass, field
21from itertools import count
22from typing import Any, TextIO
24from footman import _describe, _executor, _futures, _globals, _progress, context
25from footman._split import ChainError, Segment
26from footman.registry import (
27 Group,
28 Task,
29 _Opted,
30 cli_name,
31 default_group,
32 fans_out,
33 is_infinite,
34 is_interactive,
35 keeps_going,
36 post_deps,
37 pre_deps,
38 sharing,
39 task_confirm,
40 task_name,
41 task_retries,
42 wants_progress,
43 work_key,
44)
47@dataclass
48class _Node:
49 fn: Task
50 seg: Segment
51 key: int
52 address: str = "" # assigned once the plan is final, in creation order
53 seq: int | None = None # run-wide request stamp (the _futures counter)
54 deps: set[int] = field(default_factory=set)
55 state: str = "pending" # pending / running / done / skipped
56 result: _executor.TaskResult | None = None
57 attempts: list[_executor.TaskResult] = field(default_factory=list)
58 """The non-terminal attempts of a retried task, oldest first.
60 `result` is always the LAST attempt — the one the run's verdict, the
61 dependents and any sharer read. Every earlier attempt is a real row with
62 its own timing, output and audit, kept here and spliced into the report
63 ahead of the terminal row. Nothing is merged and nothing is hidden:
64 three attempts are three records, because each attempt *is* a record
65 (notes/20260807-timeout-and-retry.md)."""
66 forwarded: dict[str, Any] = field(default_factory=dict) # `forward`ed values in
67 forwarded_given: set[str] = field(default_factory=set) # …which were asked for
68 forward_targets: list[_Node] = field(default_factory=list) # …and out
69 keep_going: bool = False # resolved failure policy for THIS node (per-subtree)
70 shared: bool = True # resolved sharing policy: may an earlier run satisfy it?
71 # The group this default action was *reached through*, when known.
72 # Default-ness is parent-relative: a provider's default mounted into a
73 # consumer group must fan out the group it landed in, not the one it was
74 # declared on — task fns are shared between trees, so the declaration
75 # stamp cannot know where a mount placed it.
76 group: Group | None = None
79def _default_seg(fn: Task) -> Segment:
80 """The synthetic segment for a task reached by reference rather than typed —
81 a bare `pre=`/`post=` dependency, or a body call. Named the way the task is
82 *addressed* (`import_` is `fm import`), so a report never shows a spelling
83 you could not type."""
84 name = cli_name(task_name(fn))
85 return Segment(task=name, path=[name])
88def _dep_key(fn: Task) -> tuple[int, frozenset[tuple[str, Any]]]:
89 """The deduplication identity of a DAG dependency: its base task plus its
90 frozen option overrides. A bare task is simply "base with no overrides", so
91 it shares one uniform key shape with an `.opts()` reference — and an empty
92 `.opts()` (no override at all) collapses onto the bare task by construction,
93 with no int-vs-tuple asymmetry. Identical policies share a node (a shared
94 prerequisite still runs once); a genuinely different policy is a distinct
95 node. One shape with the futures memo: `registry.work_key`, sharing
96 included — a differently-shared reference is a distinct node here."""
97 return work_key(fn, include_shared=True)
100def resolve_inherited(declared: bool | None, inherited: bool) -> bool:
101 """One rung-walk for a property that flows down the dependency subtree.
103 Own declaration (or `.opts()` override) beats what the requester passed
104 down, and unset means "whoever asks decides". Sharing is the first user;
105 a later property wanting the same shape — propagate to what a task needs,
106 let a task pin its own answer — passes its own reader through here rather
107 than copying the ladder.
108 """
109 return declared if declared is not None else inherited
112def _as_task(dep: Task | Group | _Opted) -> Task:
113 """A `pre`/`post` dependency may name a runnable group; resolve it to the
114 group's default action so it runs (and fans out) like any other task. An
115 `.opts()`-wrapped group resolves the same way, carrying its overrides onto
116 the default task."""
117 if isinstance(dep, _Opted):
118 base = dep._opted_base
119 if isinstance(base, Group):
120 return _Opted(_as_task(base), dep._opted_overrides)
121 return dep # opts on a task is already a valid task reference
122 if isinstance(dep, Group):
123 if dep.default_task is None:
124 raise ChainError(
125 f"group {dep.name!r} is a prerequisite but has no @group.default"
126 )
127 return dep.default_task
128 return dep
131def _owner_of(dep: Task | Group | _Opted) -> Group | None:
132 """The group a `pre=`/`post=` reference reaches a default *through*."""
133 if isinstance(dep, _Opted):
134 base = dep._opted_base
135 return base if isinstance(base, Group) else None
136 return dep if isinstance(dep, Group) else None
139def _segment_group(root: Group, seg: Segment) -> Group | None:
140 """The group a bare-group segment names (`fm lint` — its default runs)."""
141 node = root
142 for name in seg.path:
143 sub = node.groups.get(name)
144 if sub is None:
145 return None # the path ends on a task, not a group
146 node = sub
147 return node
150def _build_dag(root: Group, segments: list[Segment]) -> list[_Node]:
151 """Nodes for the chain plus their transitive pre/post deps.
153 Explicit chain segments each get their own node, so each mention is its own
154 request (`build web build api` builds twice — different arguments, different
155 work). Shared pre/post prerequisites dedupe by task identity, so a
156 prerequisite mounted in twice is one node. Two requests that resolve to the
157 same work are one execution whether they are two nodes, a node and a body
158 call, or two calls: the second is answered by the first and reported as
159 `shared`, unless the task (or the reference) declared `shared=False`. Node
160 keys are serial ints; `dep_nodes` maps a task to the node its bare deps
161 resolve to.
162 """
163 nodes: list[_Node] = []
164 dep_nodes: dict[object, _Node] = {}
165 counter = count()
166 seen_explicit: set[object] = set()
168 def new_node(fn: Task, seg: Segment, shared: bool) -> _Node:
169 node = _Node(fn, seg, next(counter))
170 node.shared = shared
171 nodes.append(node)
172 return node
174 def add_dep(
175 fn: Task, owner: Group | None = None, *, asked_by: bool = False
176 ) -> _Node:
177 # The sharing ladder, resolved here rather than in a pass of its own:
178 # unlike `keep_going`, sharing decides node *identity*, and identity
179 # has to be settled while the node is being made. Own declaration (or
180 # `.opts()` override) wins over what the requester passed down.
181 shared = resolve_inherited(sharing(fn), asked_by)
182 if not shared:
183 # Unshared, by definition: its own node per requester, and never
184 # registered as the one a later dependency lookup could reuse.
185 node = new_node(fn, _default_seg(fn), False)
186 node.group = owner
187 _link(node)
188 return node
189 dep_node = dep_nodes.get(_dep_key(fn))
190 if dep_node is None:
191 dep_node = new_node(fn, _default_seg(fn), True)
192 dep_node.group = owner
193 dep_nodes[_dep_key(fn)] = dep_node
194 _link(dep_node)
195 return dep_node
197 splits: dict[object, _Node] = {}
199 def _thread(
200 owner: _Node, dep: _Node, fmap: dict[str, Any], fgiven: frozenset[str]
201 ) -> None:
202 # A forwarded value reaches only a dispatched task that *declares* the
203 # parameter (partial reach). Two dispatchers sending different values
204 # to a shared prerequisite are asking for different WORK: one identity
205 # rule, everywhere — requests that resolve to different arguments are
206 # different nodes, silently and correctly, never a refusal.
207 if not fmap:
208 return
209 declared = {
210 p.name for p in _executor.resolved_signature(dep.fn).parameters.values()
211 }
212 wanted = {name: value for name, value in fmap.items() if name in declared}
213 if not wanted:
214 return
215 asked = fgiven & wanted.keys()
216 # Compatible with what is already threaded: a name the dep has not seen
217 # has nothing to disagree with, and one it has must match on *both*
218 # channels — the same value asked for and merely defaulted are different
219 # requests wherever the callee reads `given()`.
220 if all(
221 name not in dep.forwarded
222 or (
223 dep.forwarded[name] == value
224 and (name in dep.forwarded_given) == (name in asked)
225 )
226 for name, value in wanted.items()
227 ):
228 dep.forwarded.update(wanted)
229 dep.forwarded_given.update(asked)
230 return
231 # The split: this dispatcher's resolved arguments name their own node.
232 # Same-argument dispatchers share the split (the plan key mirrors the
233 # execution key's normal form: declaration + frozen resolved values,
234 # plus which of them were asked for — two dispatchers sending the same
235 # value, one because it was typed and one because it was defaulted, are
236 # asking for different work wherever the callee reads `given()`).
237 frozen = tuple(
238 (name, _futures._freeze(wanted[name])) for name in sorted(wanted)
239 )
240 split_key = (_dep_key(dep.fn), frozen, frozenset(asked))
241 clone = splits.get(split_key)
242 if clone is None: 242 ↛ 258line 242 didn't jump to line 258 because the condition on line 242 was always true
243 clone = new_node(dep.fn, dep.seg, dep.shared)
244 clone.group = dep.group
245 clone.deps = set(dep.deps)
246 clone.forwarded = dict(wanted)
247 clone.forwarded_given = set(asked)
248 clone.forward_targets = list(dep.forward_targets)
249 clone.keep_going = dep.keep_going
250 splits[split_key] = clone
251 # The clone forwards onward with its own values, so its subtree
252 # sees what THIS request meant.
253 clone_map, clone_given = _executor.forward_map(
254 clone.fn, clone.seg, clone.forwarded, frozenset(clone.forwarded_given)
255 )
256 for target in clone.forward_targets: 256 ↛ 257line 256 didn't jump to line 257 because the loop on line 256 never started
257 _thread(clone, target, clone_map, clone_given)
258 if dep.key in owner.deps: # a pre: the owner now waits on the clone 258 ↛ 261line 258 didn't jump to line 261 because the condition on line 258 was always true
259 owner.deps.discard(dep.key)
260 owner.deps.add(clone.key)
261 elif owner.key in dep.deps: # a post: the clone follows the owner
262 clone.deps.add(owner.key)
263 owner.forward_targets = [
264 clone if t is dep else t for t in owner.forward_targets
265 ]
267 def _link(node: _Node) -> None:
268 """Attach *node*'s prerequisites, propagating its sharing policy down:
269 a freshly-requested task's inputs are freshly requested too, or
270 "fresh" would be a half-truth."""
271 pre = list(pre_deps(node.fn))
272 # An empty-body group default fans out the group's own tasks: they become
273 # implicit prerequisites, so the scheduler runs them (in parallel) and the
274 # default's forward-marked values thread into the ones that declare them.
275 # The `default` child is the fan-out itself — excluded from its own set.
276 group = node.group or default_group(node.fn)
277 if group is not None and fans_out(node.fn):
278 pre = [
279 *(fn for name, fn in group.tasks.items() if name != "default"),
280 *pre,
281 ]
282 for dep in pre:
283 d = add_dep(_as_task(dep), _owner_of(dep), asked_by=node.shared)
284 node.deps.add(d.key)
285 node.forward_targets.append(d) # forwarding threaded in a later pass
286 for dep in post_deps(node.fn):
287 d = add_dep(_as_task(dep), _owner_of(dep), asked_by=node.shared)
288 d.deps.add(node.key)
289 node.forward_targets.append(d)
291 for seg in segments:
292 fn = _executor.resolve(root, seg.path)
293 key = _dep_key(fn)
294 # A chain segment is a root: nothing asked for it, so only its own
295 # declaration can make it unshared.
296 shared = resolve_inherited(sharing(fn), inherited=True)
297 existing = dep_nodes.get(key) if shared else None
298 if existing is not None and key not in seen_explicit: 298 ↛ 301line 298 didn't jump to line 301 because the condition on line 298 was never true
299 # First explicit mention of a task already mounted in as a bare dep:
300 # adopt this segment's args instead of creating a duplicate.
301 existing.seg = seg
302 seen_explicit.add(key)
303 continue
304 node = new_node(fn, seg, shared)
305 node.group = _segment_group(root, seg)
306 if existing is None and node.shared:
307 dep_nodes[key] = node
308 seen_explicit.add(key)
309 _link(node)
311 # Thread forwarded values in a second pass, dependents before their deps (the
312 # reverse of the run order), so a node's *received* values are complete before
313 # it forwards on — this is what makes forwarding chain through a group default
314 # into its surfaces. It runs after segment adoption above, so each node's seg
315 # (hence its forward map) is final.
316 for node in reversed(_toposort(nodes)):
317 fmap, fgiven = _executor.forward_map(
318 node.fn, node.seg, node.forwarded, frozenset(node.forwarded_given)
319 )
320 for target in list(node.forward_targets):
321 _thread(node, target, fmap, fgiven)
322 # Addresses, once the plan is final: creation order IS request order as
323 # written (segments, then dependencies as declared, then splits), so the
324 # names are deterministic across runs and hosts.
325 labels: dict[str, int] = {}
326 for node in nodes:
327 node.address = context._next_label(labels, node.seg.task)
328 # Request stamps, in topological order and from the same run-wide counter
329 # body calls draw on: within one instant the report's tie-break then
330 # reads cause-before-consequence (a prerequisite outranks its dependent)
331 # and plan-before-body-call, deterministically across runs.
332 for node in _toposort(nodes):
333 node.seq = next(_futures._seq)
334 return nodes
337def _check_cycles(nodes: list[_Node]) -> None:
338 """Reject a cyclic dependency graph with a taught error naming the cycle.
340 Without this check the run loop would find no ready node, run nothing, and
341 exit 0 — a silent success that lies.
342 """
343 by_key = {n.key: n for n in nodes}
344 state: dict[int, int] = {} # 1 = on the current path, 2 = fully explored
346 def visit(node: _Node, path: list[str]) -> None:
347 state[node.key] = 1
348 path.append(node.seg.task)
349 for dep in node.deps:
350 child = by_key.get(dep)
351 if child is None: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 continue
353 mark = state.get(child.key, 0)
354 if mark == 1:
355 cycle = [*path[path.index(child.seg.task) :], child.seg.task]
356 raise ChainError(
357 f"dependency cycle: {' -> '.join(cycle)} "
358 f"(check the pre/post declarations of these tasks)"
359 )
360 if mark == 0:
361 visit(child, path)
362 path.pop()
363 state[node.key] = 2
365 for node in nodes:
366 if state.get(node.key, 0) == 0:
367 visit(node, [])
370def _toposort(nodes: list[_Node]) -> list[_Node]:
371 """Deps before dependents, stable by appearance order."""
372 by_key = {n.key: n for n in nodes}
373 result: list[_Node] = []
374 seen: set[int] = set()
376 def visit(node: _Node) -> None:
377 if node.key in seen:
378 return
379 seen.add(node.key)
380 for dep in node.deps:
381 if dep in by_key: 381 ↛ 380line 381 didn't jump to line 380 because the condition on line 381 was always true
382 visit(by_key[dep])
383 result.append(node)
385 for node in nodes:
386 visit(node)
387 return result
390def _plain_output(no_color: bool) -> bool:
391 """No colour at all: the `--no-color` flag, `NO_COLOR`, or a dumb terminal.
393 Per D6 this means the live rewrite is *absent*, not rewritten without escape
394 codes — the same output a pipe gets.
395 """
396 return no_color or "NO_COLOR" in os.environ or os.environ.get("TERM") == "dumb"
399def _make_ctx(
400 seg: Segment,
401 ctx_config: dict[str, Any] | None,
402 *,
403 sequential: bool,
404 capture: bool,
405 real: TextIO,
406 name_width: int = 0,
407 keep_going: bool = False,
408 shared: bool = True,
409 address: str = "",
410) -> context.Context:
411 ctx = context.Context(**(ctx_config or {}), passthrough=list(seg.passthrough or []))
412 ctx.address = address or seg.task
413 ctx.keep_going = keep_going # per-subtree policy; tags this task's subprocesses
414 # The sharing policy this node resolved to, carried so anything its body
415 # asks for inherits it: an unshared request asks unshared too.
416 ctx.shared = shared
417 # One buffer for both streams at task level: the atomic flush keeps this
418 # task's stdout/stderr in order, while a run() inside it still splits the
419 # step's streams via a temporary swap of the two.
420 ctx.sink = ctx.err_sink = None if (sequential and not capture) else io.StringIO()
421 # Step lines dress for their *destination*: a buffered block replays
422 # onto `real`, so its children style exactly as parallel() children
423 # style for their parent's terminal — both engines, one look. Only
424 # liveness (sink is None, judged in run()) gates in-place rewrites.
425 # `ansi_capable`, not bare tty-ness: the live line repaints with escape
426 # codes, and a Windows console that cannot interpret them would show
427 # the cursor dance as noise rather than a status line.
428 # Two stamps, one measurement: `terminal` is the fact (a capable terminal
429 # is at the end of this run's output, nothing captured), `tty` folds the
430 # colour policy in on top — `NO_COLOR` unsets how output *dresses*, not
431 # whether someone is watching, and the `tty()` reader answers the latter.
432 ctx.terminal = not capture and _describe.ansi_capable(real)
433 ctx.tty = ctx.terminal and not _plain_output(ctx.no_color)
434 # `--color=always` forces colour even off a terminal, but not into a captured
435 # envelope: ANSI in `--json` stdout would corrupt it, so capture wins here
436 # exactly as it does for `tty` above.
437 ctx.force_color = ctx.force_color and not capture
438 ctx.task = seg.task
439 ctx.name_width = name_width
440 return ctx
443def resolve_keep_going(root: Group, segments: list[Segment], cli: bool | None) -> bool:
444 """A run-wide *summary* of the failure policy: does anything keep going?
446 An explicit command-line choice (`-k` / `--fail-fast`) wins; unspecified,
447 true if any invoked task — a chain task or a `pre`/`post` prerequisite —
448 declares (or `.opts()`-overrides) `keep_going=True`. The scheduler resolves
449 the actual *per-node* policy with `_scope_keep_going`; this summary is for
450 callers that just want the one-bit answer.
451 """
452 if cli is not None:
453 return cli
454 try:
455 nodes = _build_dag(root, segments)
456 except (KeyError, IndexError, ChainError):
457 return False # a malformed chain surfaces its real error in run_plan
458 return any(keeps_going(n.fn) is True for n in nodes)
461def _scope_keep_going(nodes: list[_Node], cli: bool | None) -> None:
462 """Assign each node its failure policy — the per-subtree scoping.
464 A command-line `-k`/`--fail-fast` wins run-wide. Otherwise each node takes
465 its own declared (or `.opts()`-overridden) `keep_going`, and a keep-going
466 node propagates that down its own subtree — its `pre`/`post` prerequisites
467 keep going with it — so a mixed chain honours each side: a keep-going gate
468 surfaces all of its own failures while an independent fail-fast task still
469 bails on the first. A node's own policy always wins over an inherited one, so
470 an explicit fail-fast prerequisite stays a fail-fast boundary. Unspecified
471 everywhere is the built-in fail-fast.
472 """
473 if cli is not None:
474 for node in nodes:
475 node.keep_going = cli
476 return
477 inherited: set[int] = set() # keys a keep-going dependent has reached
478 for node in reversed(_toposort(nodes)): # dependents resolve before their deps
479 own = keeps_going(node.fn) # True / False / None (reads declared + opted)
480 node.keep_going = own if own is not None else node.key in inherited
481 if node.keep_going:
482 inherited.update(node.deps) # keep this subtree's prerequisites going
485def dag_wants_progress(root: Group, segments: list[Segment]) -> bool:
486 """Whether every task in the expanded DAG — pre/post deps included —
487 consented to timing. One `@task(progress=False)` opts the run out of
488 recording and of a determinate bar (the pulse still shows)."""
489 try:
490 nodes = _build_dag(root, segments)
491 except (KeyError, IndexError, ChainError):
492 return False # a malformed chain surfaces its real error in run_plan
493 return all(wants_progress(n.fn) for n in nodes)
496class NotConfirmed(Exception):
497 """A `@task(confirm=…)` gate was declined (or unanswerable off a terminal)."""
499 def __init__(self, task: str) -> None:
500 super().__init__(f"{task}: not confirmed")
503def _ask_confirm(message: str, *, no_input: bool) -> bool:
504 """The `@task(confirm=)` gate. Off a terminal or under `--no-input` the
505 answer is no — like just and go-task, a confirm fails without `--yes`
506 rather than proceeding unasked. Asked on stderr before output routing."""
507 if no_input or not context._stdin_is_tty():
508 return False
509 reply = context._prompt_core(f"{message} [y/N] ", default="n")
510 return reply.strip().lower() in ("y", "yes")
513def _gate_confirms(
514 root: Group, segments: list[Segment], ctx_config: dict[str, Any] | None
515) -> tuple[list[Segment], list[_executor.TaskResult], dict[Any, bool]]:
516 """Resolve each invoked task's `@task(confirm=)` before the DAG is built —
517 asked in invocation order, before any prerequisite runs. A confirmed task
518 is kept; a denied one is dropped (so its exclusive pre-deps are pruned with
519 it) and reported as a failed 'not confirmed' result, so the run exits
520 non-zero. `--yes` auto-confirms every gate. One reference, one question:
521 the answers come back so the node-level gate (prerequisites, fan-out
522 members) never re-asks what the invocation already answered."""
523 cfg = ctx_config or {}
524 assume_yes = bool(cfg.get("assume_yes"))
525 no_input = bool(cfg.get("no_input"))
526 dry_run = bool(cfg.get("dry_run"))
527 kept: list[Segment] = []
528 denied: list[_executor.TaskResult] = []
529 answers: dict[Any, bool] = {}
530 for seg in segments:
531 fn = _executor.resolve(root, seg.path)
532 message = task_confirm(fn)
533 if not message:
534 kept.append(seg)
535 continue
536 key = _dep_key(fn)
537 if key not in answers: 537 ↛ 545line 537 didn't jump to line 545 because the condition on line 537 was always true
538 answers[key] = _answer_confirm(
539 seg.task,
540 message,
541 assume_yes=assume_yes,
542 no_input=no_input,
543 dry_run=dry_run,
544 )
545 if answers[key]:
546 kept.append(seg)
547 else:
548 denied.append(_not_confirmed(seg))
549 return kept, denied, answers
552def _answer_confirm(
553 task: str, message: str, *, assume_yes: bool, no_input: bool, dry_run: bool
554) -> bool:
555 """The one confirm ladder, whichever moment asks — a segment gate, a
556 node gate, a body call: `--yes` answers first, a rehearsal assumes yes
557 out loud, and only then is a human asked. Three sites used to spell it
558 separately."""
559 return (
560 assume_yes
561 or (dry_run and _dry_confirm(task, message))
562 or _ask_confirm(message, no_input=no_input)
563 )
566def _dry_confirm(task: str, message: str) -> bool:
567 """A rehearsal answers every gate yes — a gate answered no would hide
568 the very work the rehearsal exists to show — and says so, once."""
569 from footman import _globals
571 _globals._note(f"dry-confirm:{task}", f"dry-run: {message!r} — assumed yes")
572 return True
575def _gate_node_confirms(
576 nodes: list[_Node],
577 ctx_config: dict[str, Any] | None,
578 answers: dict[Any, bool],
579) -> None:
580 """Ask the plan's remaining `@task(confirm=)` gates — prerequisites and
581 fan-out members — up front, in dependency order: a task that asks for
582 confirmation gets it however it was reached. One reference, one question;
583 an answer given for a segment covers every node resolving to the same
584 reference. A denial becomes the node's result before anything runs, so
585 the node never launches and its dependents skip with the denial as their
586 cause."""
587 cfg = ctx_config or {}
588 assume_yes = bool(cfg.get("assume_yes"))
589 no_input = bool(cfg.get("no_input"))
590 dry_run = bool(cfg.get("dry_run"))
591 for n in _toposort(nodes):
592 message = task_confirm(n.fn)
593 if not message or n.result is not None:
594 continue
595 key = _dep_key(n.fn)
596 if key not in answers:
597 answers[key] = _answer_confirm(
598 n.seg.task,
599 message,
600 assume_yes=assume_yes,
601 no_input=no_input,
602 dry_run=dry_run,
603 )
604 if not answers[key]:
605 n.result = _not_confirmed(n.seg)
606 n.result.address = n.address
607 n.result.seq = n.seq
608 n.state = "done"
611def _not_confirmed(seg: Segment) -> _executor.TaskResult:
612 # Denied at a moment (before the run for a segment, at the call for a call),
613 # so it carries that moment and needs no special placement.
614 return _executor.TaskResult(
615 task=seg.task,
616 ok=False,
617 code=1,
618 error=NotConfirmed(seg.task),
619 started=time.perf_counter(),
620 )
623def _in_request_order(
624 results: list[_executor.TaskResult],
625) -> list[_executor.TaskResult]:
626 """The run's results in the order the work was created.
628 Request order has a place for everything: a task reached by a body call
629 has no slot in a dependency listing, but it drew a stamp from the same
630 run-wide counter the plan did, at the moment it was asked for. Sequential
631 runs are unchanged, since requesting in dependency order *is* the order
632 they run in.
634 It is also the only ordering a reader can rely on. The clock cannot do
635 this job: two independent tasks in a parallel run start in whatever order
636 the pool hands them workers, so a report keyed on `started` reshuffles
637 between runs of the same command — and did, intermittently, on a
638 free-threaded build. Sorting by start time within a 10ms bucket hid it
639 for pairs that shared a bucket and left the pair that straddled a
640 boundary to chance, because the buckets are aligned to `perf_counter`'s
641 arbitrary origin rather than to anything about the run.
643 Something that never began has no stamp of its own, so it sits directly
644 after whatever prevented it — the report reads as cause, then consequence.
645 With nothing to blame (a gate answered before any task ran) it comes first.
646 """
647 # `started` breaks a tie only for a row minted outside the request
648 # pipeline, which carries no stamp to sort by.
649 ran = sorted(
650 (r for r in results if r.started is not None),
651 key=lambda r: (
652 r.seq if r.seq is not None else float("inf"),
653 r.started or 0.0,
654 ),
655 )
656 never: list[_executor.TaskResult] = [r for r in results if r.started is None]
657 ordered = [r for r in never if not r.blocked_by] + ran
658 for result in (r for r in never if r.blocked_by):
659 after = [i for i, r in enumerate(ordered) if r.task == result.blocked_by]
660 ordered.insert(after[-1] + 1 if after else len(ordered), result)
661 return ordered
664def confirm_gate(
665 fn: Task, seg: Segment, ctx: context.Context
666) -> _executor.TaskResult | None:
667 """Ask *fn*'s `@task(confirm=)` gate now; a denial is the refusal to report.
669 The scheduler resolves a segment's gate before the run starts, so the human
670 answers everything up front. A body call cannot be known that early, so it
671 asks at the moment of the call instead — the gate itself is the same, and a
672 task that asks for confirmation gets it however it was reached.
673 """
674 message = task_confirm(fn)
675 if not message:
676 return None
677 if _answer_confirm(
678 seg.task,
679 message,
680 assume_yes=ctx.assume_yes,
681 no_input=ctx.no_input,
682 dry_run=ctx.dry_run,
683 ):
684 return None
685 return _not_confirmed(seg)
688def run_plan(
689 root: Group,
690 segments: list[Segment],
691 *,
692 sequential: bool = False,
693 keep_going: bool | None = None,
694 capture: bool = False,
695 ctx_config: dict[str, Any] | None = None,
696 estimate: _progress.Estimate | None = None,
697 progress: bool = True,
698 jobs: int = 0,
699) -> list[_executor.TaskResult]:
700 """Build and run the DAG; return results in dependency order."""
701 with _futures.session():
702 results = _run_plan(
703 root,
704 segments,
705 sequential=sequential,
706 keep_going=keep_going,
707 capture=capture,
708 ctx_config=ctx_config,
709 estimate=estimate,
710 progress=progress,
711 jobs=jobs,
712 )
713 # A task reached by a body call ran as a real task, so its result joins
714 # the run's. Every execution the run performed is reported, however it
715 # was reached — and the whole report reads in the order it happened.
716 return _in_request_order([*results, *_futures.collected()])
719def _run_plan(
720 root: Group,
721 segments: list[Segment],
722 *,
723 sequential: bool = False,
724 keep_going: bool | None = None,
725 capture: bool = False,
726 ctx_config: dict[str, Any] | None = None,
727 estimate: _progress.Estimate | None = None,
728 progress: bool = True,
729 jobs: int = 0,
730) -> list[_executor.TaskResult]:
731 context.reset_abort() # clear any latched fail-fast from a previous run
732 segments, denied, answers = _gate_confirms(root, segments, ctx_config)
733 nodes = _build_dag(root, segments)
734 _check_cycles(nodes)
735 _gate_node_confirms(nodes, ctx_config, answers)
736 _scope_keep_going(nodes, keep_going) # per-node failure policy (tri-state + scope)
737 # Ask-serial, run-parallel — as early as correct: every promptable ask()
738 # across the DAG answers up front (confirms were gated above), so the
739 # human answers once and walks away; only a live-suggest question waits
740 # for its prerequisites, resolving at node launch. Think-time can never
741 # land inside any task's recorded duration, and an unanswerable question
742 # (--no-input, no terminal) refuses the run before anything starts.
743 ask_ctx = context.Context(**(ctx_config or {}))
744 try:
745 for n in _toposort(nodes):
746 if n.result is not None: # denied its confirm: it will not run
747 continue
748 _executor.resolve_asks(n.fn, n.seg, ask_ctx)
749 except ValueError as exc:
750 raise ChainError(str(exc)) from exc
751 # One node has nothing to parallelise — run it on the sequential-live
752 # path instead: output streams as it happens, and run()'s TTY mode
753 # (colour, in-place step rewrite) applies. `fm check` is this shape. An
754 # interactive task also forces sequential: it owns the terminal, so it
755 # can't share with parallel siblings (a human-wait is the bottleneck).
756 # An interactive task owns the real terminal: it forces sequential (it can't
757 # share with parallel siblings) and suppresses the status line, whose
758 # clear-line repaints would otherwise erase its prompt.
759 # An interactive task neither forces the run sequential nor costs it
760 # the status line: it claims the arbiter's console lane (one owner,
761 # real stdio), the parallel pool keeps running around it captured, and
762 # the lane suspends the status line for exactly the ownership window.
763 sequential = sequential or len(nodes) == 1
764 # A run containing an infinite task has no progress to show — its
765 # duration isn't late, it's intentional. The status line yields to a
766 # one-time hint (printed at the node's start) saying how this ends.
767 endless = any(is_infinite(n.fn) for n in nodes)
768 # Resolve colour once for the whole run and publish it into os.environ, so
769 # every tool — subprocess (inherits) and in-process (reads it) — sees one
770 # answer set once, with no per-call environment patching. sys.stdout is the
771 # run's real stdout here (routing captures it next), so its tty-ness is the
772 # same input `_make_ctx` uses per node.
773 cfg = ctx_config or {}
774 try:
775 stdout_tty = sys.stdout.isatty()
776 except Exception:
777 stdout_tty = False
778 colour_on = context.run_colour_on(
779 no_color=bool(cfg.get("no_color")),
780 force_color=bool(cfg.get("force_color")),
781 capture=capture,
782 isatty=stdout_tty,
783 )
784 with context.routing() as (real, err):
785 # Decide the status line and the infinite-task hint from the *user's*
786 # environment — before `color_environment` publishes footman's own
787 # NO_COLOR/FORCE_COLOR for the children, which `_plain_output` would
788 # otherwise read back and mistake for a user's monochrome request (a
789 # `fm check > log` still wants its spinner on the stderr terminal).
790 status = _make_status(
791 err,
792 ctx_config,
793 capture,
794 estimate,
795 # An interactive run keeps its status line now: the console
796 # lane suspends it while a wizard owns the terminal and resumes
797 # it after — instead of sacrificing it for the whole run.
798 progress and not endless,
799 )
800 hint_err = (
801 err
802 if sequential
803 and endless
804 and not capture
805 and not cfg.get("quiet")
806 and err.isatty()
807 and not _plain_output(bool(cfg.get("no_color")))
808 else None
809 )
810 if status is not None:
811 status.unit_added(sum(1 for n in nodes if n.result is None))
812 context.set_status(status) # parallel() and the routers find it
813 status.open()
814 try:
815 with context.color_environment(colour_on):
816 # The process-globals routers arm inside the colour publish,
817 # so the pinned env snapshot carries it; refcounted, so a
818 # nested run (a task body driving a Runner) shares one install.
819 _globals.install()
820 try:
821 if sequential:
822 try:
823 _run_sequential(
824 nodes, real, capture, ctx_config, status, hint_err
825 )
826 except BaseException:
827 # Ctrl-C mid-task: the running child is group-isolated,
828 # so it missed the terminal's SIGINT — reap its tree by
829 # hand before the interrupt propagates.
830 context.terminate_live_children()
831 raise
832 else:
833 _run_parallel(
834 nodes, real, err, capture, ctx_config, status, jobs
835 )
836 finally:
837 _globals.uninstall()
838 finally:
839 if status is not None:
840 context.set_status(None)
841 status.close()
842 # The abort latch is run-scoped: leaving it set after a normal return
843 # would make `_register_child` reap a *later* bare run()'s child (a
844 # latched fail-fast from one test killing the next test's echo). The
845 # Ctrl-C/internal-error path unwinds by exception and keeps the latch,
846 # which the interrupted-reporting above this layer still reads.
847 context.reset_abort()
848 ordered = _toposort(nodes)
849 by_key = {n.key: n for n in ordered}
850 already = {r.task for r in denied}
852 def _lost(key: int) -> bool:
853 # Judged on final states, after the run: a dependency is the cause
854 # whether it failed having completed, was cancelled mid-flight, was
855 # itself skipped — or never ran under a name the run denied (its
856 # refusal is a pre-run row, so its node reads as merely pending;
857 # the result-None guard keeps a same-named node that *did* run from
858 # being blamed for it).
859 m = by_key.get(key)
860 return m is not None and (
861 m.state == "skipped"
862 or (m.result is None and m.seg.task in already)
863 or (m.result is not None and not m.result.ok)
864 )
866 def _blocked_by(n: _Node) -> str:
867 for dep in n.deps:
868 if _lost(dep):
869 return by_key[dep].seg.task
870 # A fail-fast sweep: nothing of its own was lost — blame the run's
871 # first genuine failure, the same cause the exit code reports.
872 for m in ordered: 872 ↛ 875line 872 didn't jump to line 875 because the loop on line 872 didn't complete
873 if m.result is not None and not m.result.ok:
874 return m.seg.task
875 return ""
877 # A node the run never started still gets a row — `skipped`, blamed on
878 # what prevented it — so the report and the `--json` envelope account
879 # for every node the plan had, not only the ones that ran. Marked
880 # `skipped` by a sweep, or simply left pending when the run stopped
881 # reaching for new work: either way it never began, so it has no moment
882 # of its own and `_in_request_order` seats it after its cause. A node whose
883 # confirm was denied already has its refusal row.
884 for n in ordered:
885 # Launch latency, derivable after the fact: a node became eligible
886 # the moment its last prerequisite finished — same monotonic clock
887 # `started` reads, so `started - eligible` is the time it sat
888 # waiting for a worker. Roots have no prerequisites and no latency.
889 if n.result is None or n.result.started is None or not n.deps:
890 continue
891 # The edges themselves, by row address — what a profile draws its
892 # dependency arrows from, and the `after` a `--json` reader gets.
893 n.result.after = tuple(
894 m.result.address or m.seg.task
895 for d in n.deps
896 if (m := by_key.get(d)) is not None and m.result is not None
897 )
898 finishes = [
899 m.result.started + m.result.duration
900 for d in n.deps
901 if (m := by_key.get(d)) is not None
902 and m.result is not None
903 and m.result.started is not None
904 ]
905 if finishes: 905 ↛ 884line 905 didn't jump to line 884 because the condition on line 905 was always true
906 n.result.eligible = max(finishes)
907 skipped = [
908 _executor.TaskResult(
909 task=n.seg.task,
910 # Addresses are assigned when the plan is final, not when a task
911 # runs — so a row that never ran still has one, and `blocked_by`
912 # and `after` point at names a reader can look up.
913 address=n.address,
914 ok=False,
915 state="skipped",
916 blocked_by=_blocked_by(n),
917 )
918 for n in ordered
919 if n.result is None and n.seg.task not in already
920 ]
921 # A retried task contributes its earlier attempts ahead of its terminal
922 # row, so the report reads in the order the work happened: attempt, then
923 # attempt, then the outcome. They are ordinary rows — real timing, real
924 # output, their own audit — distinguished only by `state="retried"`.
925 ran_rows: list[_executor.TaskResult] = []
926 for n in ordered:
927 if n.result is None:
928 continue
929 for earlier in n.attempts:
930 # One request, so one stamp — the sort then falls to `started`,
931 # which is chronological because a node's attempts run in
932 # sequence on one thread.
933 if earlier.seq is None: 933 ↛ 935line 933 didn't jump to line 935 because the condition on line 933 was always true
934 earlier.seq = n.result.seq
935 ran_rows.append(earlier)
936 ran_rows.append(n.result)
937 return denied + ran_rows + skipped
940def _run_sequential(
941 nodes: list[_Node],
942 real: TextIO,
943 capture: bool,
944 ctx_config: dict[str, Any] | None,
945 status: _progress.StatusLine | None,
946 err: TextIO | None = None,
947) -> None:
948 done: dict[int, bool] = {}
949 failed = False
950 width = max((_describe.display_width(n.seg.task) for n in nodes), default=0)
951 for node in _toposort(nodes):
952 if node.result is not None: # answered before the run: a denied confirm 952 ↛ 953line 952 didn't jump to line 953 because the condition on line 952 was never true
953 done[node.key] = node.result.ok
954 failed = failed or not node.result.ok
955 continue
956 if any(not done.get(d) for d in node.deps) or (failed and not node.keep_going):
957 node.state = "skipped"
958 if status is not None: 958 ↛ 959line 958 didn't jump to line 959 because the condition on line 958 was never true
959 status.unit_skipped(node.seg.task)
960 continue
961 ctx = _make_ctx(
962 node.seg,
963 ctx_config,
964 sequential=True,
965 capture=capture,
966 real=real,
967 name_width=width,
968 keep_going=node.keep_going,
969 shared=node.shared,
970 address=node.address,
971 )
972 if status is not None:
973 status.unit_started(node.seg.task)
974 if err is not None and is_infinite(node.fn):
975 hint = f"{node.seg.task} runs until you stop it — Ctrl-C"
976 err.write(_describe.dim(hint, True) + "\n")
977 err.flush()
978 node.result = _run_attempts(node, ctx)
979 if node.result.seq is None or node.seq is None:
980 node.result.seq = node.seq
981 else: # a shared row keeps its above-the-record floor
982 node.result.seq = max(node.result.seq, node.seq)
983 node.state = "done"
984 if status is not None:
985 status.unit_finished(node.seg.task, node.result.ok)
986 done[node.key] = node.result.ok
987 failed = failed or not node.result.ok
990def _run_attempts(node: _Node, ctx: Any) -> _executor.TaskResult:
991 """Run one node's body, retrying a failed attempt while attempts remain.
993 Every attempt is a real record. The non-terminal ones are kept on the
994 node (`attempts`) and reported as their own `retried` rows; the last one
995 becomes `node.result` — the row the run's verdict, the dependents and any
996 sharer read. Nothing is merged, so `records are never fiction` holds
997 without special pleading: each attempt IS a record.
999 The retry lives here, above `run_task`, for the property ruling 1 needs:
1000 a retriable failure never reaches the scheduler's failure handling, so it
1001 latches no fail-fast and blocks no dependent. There is no failure to
1002 react to until the attempts are spent — *"it hasn't failed yet"*.
1004 What is NOT re-run: `pre=` (prerequisites are their own nodes and already
1005 ran), and every gate that guards an attempt rather than performing it —
1006 availability, `expose`, and above all the confirm prompt, which
1007 resolved before this is called. A retry that re-prompts a human is a bug
1008 read as broken rather than as an oversight.
1010 Fail-fast still wins over a pending retry: an abort latched by a
1011 *different* task's terminal failure means "no new work", and an unstarted
1012 attempt is new work.
1014 **A timeout footman could not stop is terminal**, whatever attempts
1015 remain. With `stopped=True` the worst a retry costs is a repeated side
1016 effect: attempt 1 finished doing whatever it did, then attempt 2 does it
1017 again. With `stopped=False` the body is *still running*, so attempt 2
1018 would race a live copy of itself — two writers on one file, two calls
1019 against one API, at the same time. That is a fork, not a retry.
1021 The practical cost is worse than the semantic one: `retries=3` on a body
1022 with no checkpoints leaks three hung workers, which under a bounded
1023 `--jobs` exhausts the pool and can wedge the run at exit on non-daemon
1024 threads. And it cannot succeed anyway — a body that blew its deadline
1025 without reaching a checkpoint has no checkpoint to reach, so the next
1026 attempt hangs identically.
1028 This does not contradict ruling 3 ("no theory about what deserves
1029 retry"). That forbids footman judging whether a *failure* deserves
1030 another chance; this is footman observing it cannot coherently *start*
1031 another attempt, because the previous one never ended — the same
1032 category as fail-fast beating a pending retry. Deliberately not
1033 configurable: a flag would be API surface for a case unsafe by
1034 construction (notes/20260807-timeout-and-retry.md).
1035 """
1036 left = task_retries(node.fn)
1037 attempt = 0
1038 while True:
1039 result = _executor.run_task(
1040 node.fn,
1041 node.seg,
1042 ctx,
1043 node.forwarded,
1044 frozenset(node.forwarded_given),
1045 attempt=attempt,
1046 )
1047 if result.ok or left <= 0:
1048 return result
1049 if result.timed_out and result.after_deadline != "stopped":
1050 # Only a *stopped* timeout is retriable — cut off mid-flight, so
1051 # possibly slow for a transient reason and definitively not
1052 # running now. A `completed` one is terminal through futility:
1053 # the body finished, so there is nothing transient to retry, it
1054 # outran the deadline once and will again, and another attempt
1055 # repeats work that already happened.
1056 #
1057 # This is not footman forming a theory about which failures
1058 # deserve another chance (ruling 3) — it is a structural fact
1059 # about whether another attempt can coherently start, the same
1060 # category as fail-fast beating a pending retry.
1061 return result
1062 result.state = "retried"
1063 node.attempts.append(result)
1064 left -= 1
1065 attempt += 1
1066 # Each attempt keeps the node's request stamp — they are one request —
1067 # so the report's `(seq, started)` sort would place them by start time
1068 # alone. That is already chronological, but only because every attempt
1069 # of one node runs on one thread; pinning the stamp here says so
1070 # deliberately rather than relying on it.
1073def _make_status(
1074 err: TextIO,
1075 ctx_config: dict[str, Any] | None,
1076 capture: bool,
1077 estimate: _progress.Estimate | None,
1078 enabled: bool,
1079) -> _progress.StatusLine | None:
1080 """The run's live line — bar or pulse — or None when it can't show.
1082 Status is commentary, so it lives on stderr: piping stdout
1083 (`fm check > log`) keeps the line visible on the terminal. Applies to
1084 every run shape, single node included — that's `fm check`.
1085 """
1086 cfg = ctx_config or {}
1087 if (
1088 not enabled
1089 or capture
1090 or cfg.get("quiet")
1091 or not err.isatty() # the status stream's own tty-ness decides
1092 or _plain_output(bool(cfg.get("no_color")))
1093 ):
1094 return None
1095 # Past the guard the run is colourful by definition (no_color/NO_COLOR/dumb
1096 # all bail above), so the live line always renders with escapes.
1097 return _progress.StatusLine(err, estimate, color=True)
1100def _run_parallel(
1101 nodes: list[_Node],
1102 real: TextIO,
1103 err: TextIO | None,
1104 capture: bool,
1105 ctx_config: dict[str, Any] | None,
1106 status: _progress.StatusLine | None,
1107 jobs: int,
1108) -> None:
1109 by_key = {n.key: n for n in nodes}
1110 lock = threading.Lock()
1111 # A confirm denied before the run is already a failure on the board.
1112 failed = any(n.result is not None and not n.result.ok for n in nodes)
1113 width = max((_describe.display_width(n.seg.task) for n in nodes), default=0)
1115 def dep_ok(n: _Node) -> bool:
1116 return all(
1117 by_key[d].state == "done"
1118 and (res := by_key[d].result) is not None
1119 and res.ok
1120 for d in n.deps
1121 if d in by_key
1122 )
1124 def dep_lost(n: _Node) -> bool:
1125 def lost(m: _Node) -> bool:
1126 return m.state == "skipped" or (
1127 m.state == "done" and (res := m.result) is not None and not res.ok
1128 )
1130 return any(lost(by_key[d]) for d in n.deps if d in by_key)
1132 def run_node(n: _Node) -> None:
1133 ctx = _make_ctx(
1134 n.seg,
1135 ctx_config,
1136 sequential=False,
1137 capture=capture,
1138 real=real,
1139 name_width=width,
1140 keep_going=n.keep_going,
1141 shared=n.shared,
1142 address=n.address,
1143 )
1144 if is_interactive(n.fn) and not capture:
1145 # A console owner runs on the real terminal even inside the
1146 # parallel pool: the arbiter's console lane guarantees one owner,
1147 # and captured siblings' flushes queue on the gate below.
1148 ctx.sink = ctx.err_sink = None
1149 n.result = _run_attempts(n, ctx)
1150 if n.result.seq is None or n.seq is None:
1151 n.result.seq = n.seq
1152 else: # a shared row keeps its above-the-record floor
1153 n.result.seq = max(n.result.seq, n.seq)
1154 if not capture and ctx.sink is not None:
1155 # Flush this task's buffered output as one block — queued while a
1156 # wizard owns the terminal, so it never splats over a prompt.
1157 with _globals.console_gate(), lock:
1158 # The per-task sink is always the StringIO buffer set above;
1159 # the isinstance states the invariant where the type (TextIO)
1160 # can't.
1161 sink = ctx.sink
1162 blob = sink.getvalue() if isinstance(sink, io.StringIO) else ""
1163 if status is not None:
1164 # A direct real-stream write (bypasses the routers): the
1165 # status line clears itself and tracks the column.
1166 status.notify(blob)
1167 real.write(blob)
1168 real.flush()
1170 # A one-node plan has no sibling to overlap with, so the pool would be a
1171 # thread to wait on and nothing to gain. Run it on this thread instead and
1172 # skip the import below entirely — the single most common shape (`fm test`,
1173 # a bare task with no prerequisites) stops paying `concurrent.futures`.
1174 # The *regime* is unchanged: run_node still builds a parallel context, so a
1175 # task behaves the same whether it was named alone or in a chain.
1176 if len(nodes) == 1: 1176 ↛ 1177line 1176 didn't jump to line 1177 because the condition on line 1176 was never true
1177 node = nodes[0]
1178 if node.result is None and not failed:
1179 node.state = "running"
1180 if status is not None:
1181 status.unit_started(node.seg.task)
1182 try:
1183 run_node(node)
1184 except BaseException:
1185 # Same reap as the pool's abort path: the child is
1186 # group-isolated and missed the terminal's SIGINT.
1187 context.terminate_live_children()
1188 raise
1189 node.state = "done"
1190 if status is not None:
1191 status.unit_finished(
1192 node.seg.task, bool(node.result and node.result.ok)
1193 )
1194 return
1196 # Imported at the pool, not at module scope: `concurrent.futures` costs
1197 # ~5.9 ms (it drags `logging` through `traceback`), and a line that never
1198 # runs a plan — `--list`, `--help`, a refusal — should not pay it.
1199 from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
1201 with ThreadPoolExecutor(
1202 max_workers=jobs if jobs > 0 else None, thread_name_prefix="fm-worker"
1203 ) as pool:
1204 futures: dict[Any, _Node] = {}
1205 try:
1206 while True:
1207 for n in nodes:
1208 if n.state == "pending" and (
1209 dep_lost(n) or (failed and not n.keep_going)
1210 ):
1211 n.state = "skipped"
1212 if status is not None: 1212 ↛ 1213line 1212 didn't jump to line 1213 because the condition on line 1212 was never true
1213 status.unit_skipped(n.seg.task)
1214 for n in nodes:
1215 if n.state == "pending" and dep_ok(n):
1216 n.state = "running"
1217 if status is not None:
1218 status.unit_started(n.seg.task)
1219 futures[pool.submit(run_node, n)] = n
1220 if not futures:
1221 break
1222 completed, _ = wait(list(futures), return_when=FIRST_COMPLETED)
1223 for fut in completed:
1224 node = futures.pop(fut)
1225 # `run_task` catches task exceptions itself; anything the
1226 # future carries (KeyboardInterrupt in the worker, an
1227 # internal error) must propagate, not read as success.
1228 exc = fut.exception()
1229 if exc is not None: 1229 ↛ 1230line 1229 didn't jump to line 1230 because the condition on line 1229 was never true
1230 raise exc
1231 node.state = "done"
1232 ok = bool(node.result and node.result.ok)
1233 if status is not None:
1234 status.unit_finished(node.seg.task, ok)
1235 if not ok:
1236 failed = True
1237 # True fail-fast: stop launching new nodes (the skip pass
1238 # above) *and* reap the FAIL-FAST siblings already in
1239 # flight, so a doomed branch dies now instead of waiting
1240 # out a five-minute test suite. `failfast_only` spares a
1241 # keep-going task in a mixed run — it isn't doomed.
1242 context.terminate_live_children(failfast_only=True)
1243 except BaseException:
1244 # Abort (Ctrl-C, or an internal error surfaced above): drop
1245 # everything not yet started, then kill in-flight subprocess trees.
1246 # This must happen *before* the pool's `with` exit joins the worker
1247 # threads: each is blocked in communicate() on a group-isolated child
1248 # that no longer receives the terminal's SIGINT, so without an
1249 # explicit kill the join — and the whole Ctrl-C — would hang. The app
1250 # layer reports "interrupted" and exits 130; run_plan's finally
1251 # clears the status line.
1252 context.terminate_live_children()
1253 pool.shutdown(wait=False, cancel_futures=True)
1254 raise